﻿using System;
using System.Collections;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.Diagnostics;
using System.IO;
using System.Reflection;
using System.Resources;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading;
using Common.Utilities;
using Common.Utilities.Dispatcher;
using Workflows.Components.Entities.Distances.Comparers;

namespace Workflows.Components.Entities.Collapses
{
    /// <summary>
    /// provide a realtime callback to update charts
    /// </summary>
    /// <param name="entityNotCollapsed"></param>
    /// <param name="entityCollapsed"></param>
    /// <param name="collapsedEntities"></param>
    public delegate void CollapseProgressing(int entityNotCollapsed, int entityCollapsed, int collapsedEntities);

    /// <summary>
    /// 
    /// </summary>
    /// <param name="fieldContributions"></param>
    public delegate void FieldsContributionUpdated(
        Dictionary<string, DescriptiveAttributeContribution> fieldContributions);

    /// <summary>
    /// 
    /// </summary>
    /// <param name="dtEntities"></param>
    public delegate void UnderCollapsedEntitiesUpdated(DataTable dtEntities);

    /// <summary>
    /// 
    /// </summary>
    /// <param name="dtUnderCollapsedEntities"></param>
    /// <param name="collapsedEntitiesTableName"></param>
    /// <param name="collapsedIDFieldName"></param>
    public delegate void UnderCollapsedEntitiesPopulated(DataTable dtUnderCollapsedEntities, string collapsedEntitiesTableName, string collapsedIDFieldName);

    /// <summary>
    /// 
    /// </summary>
    /// <param name="dtEntities"></param>
    public delegate void OverCollapsedEntitiesUpdated(DataTable dtEntities);

    /// <summary>
    /// 
    /// </summary>
    /// <param name="dtOverCollapsedEntities"></param>
    /// <param name="collapsedEntitiesTableName"></param>
    /// <param name="collapsedIDFieldName"></param>
    public delegate void OverCollapsedEntitiesPopulated(DataTable dtOverCollapsedEntities, string collapsedEntitiesTableName, string collapsedIDFieldName);

    /// <summary>
    /// 
    /// </summary>
    public struct EntityCompareResult
    {
        public string FieldName;
        public string FieldValueA;
        public string FieldValueB;
        public double FieldProbability;
        public double FieldWeight;
        public double FieldCompareScore;
    }

    /// <summary>
    /// 
    /// </summary>
    public class EntityCollapseWorker:IWorker
    {
        public const string CollapseIDFieldName = "CollapsedID";

        private Thread _WorkerThread;
        private EntityCollapseInstruction _Instruction;
        private EntityCollapseOutput _Output;
        private Guid _JobTicketID;
        private RunContext _Context;
        private string _WorkflowName;
        private string _ComponentName;

        /// <summary>
        /// 
        /// </summary>
        public string ComponentName
        {
            get { return _ComponentName; }
        }

        /// <summary>
        /// 
        /// </summary>
        public string WorkflowName
        {
            get { return _WorkflowName; }
        }

        private const string PopulateFieldValueDistributionStoredProcName = "PopulateKeyPersonnelFieldValueDistributionStats";
        private int _LastCollapseID = 0;

        // test mode events
        public event FieldsContributionUpdated OnFieldContributionsUpdated;
        public event UnderCollapsedEntitiesUpdated OnUnderCollapsedEntitiesUpdated;
        public event UnderCollapsedEntitiesPopulated OnUnderCollapsedEntitiesPopulated;
        public event OverCollapsedEntitiesUpdated OnOverCollapsedEntitiesUpdated;
        public event OverCollapsedEntitiesPopulated OnOverCollapsedEntitiesPopulated;
        public event CollapseProgressing OnCollapseProgressing;
        private bool _TestMode;
        /// <summary></summary>
        public bool TestMode
        {
            get
            {
                return this._TestMode;
            }
            set
            {
                this._TestMode = value;
            }
        }
        private int _TableSampleSize = 1000;
        /// <summary></summary>
        public int TableSampleSize
        {
            get
            {
                return this._TableSampleSize;
            }
            set
            {
                this._TableSampleSize = value;
            }
        }

        #region IWorker Members

        public Thread WorkerThread
        {
            get { return _WorkerThread; }
        }

        public InstructionBase Instruction
        {
            get { return _Instruction; }
        }

        public ResultBase Output
        {
            get { return _Output; }
        }

        public Type ComponentType
        {
            get { return typeof(EntityCollapseComponent); }
        }

        public RunContext Context
        {
            get { return _Context; }
            set { _Context = value; }
        }

        public void SetInstruction(Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._JobTicketID = ticketGuid;
            this._Instruction = (EntityCollapseInstruction) request;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }

        public void SetInstruction(RunContext context, Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._JobTicketID = ticketGuid;
            this._Instruction = (EntityCollapseInstruction)request;
            this._Context = context;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }

        public void DoWork()
        {
            if(this.WorkerStartedEvent !=null)
            {
                this.WorkerStartedEvent(this._JobTicketID, this, this.GetType(), this._Instruction);
            }
            this._WorkerThread = Thread.CurrentThread;
            this.CollapseEntities();
        }

        public void DoWorkAsync()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._JobTicketID, this, this.GetType(), this._Instruction);
            }
            this._WorkerThread = new Thread(new ThreadStart(this.CollapseEntities));
            this._WorkerThread.Start();
        }

        public void CancelWork()
        {
            if(this._WorkerThread !=null && this._WorkerThread.IsAlive)
            {
                this._WorkerThread.Abort();
                if (this.WorkerCancelledEvent != null)
                {
                    this.WorkerCancelledEvent(this._JobTicketID, this.GetType(), this._Instruction);
                }
            }
            
        }

        public event WorkerStarted WorkerStartedEvent;
        public event WorkerFinished WorkerFinishedEvent;
        public event WorkerHasError WorkerErredEvent;
        public event WorkerCancelled WorkerCancelledEvent;
        public event WorkerStatusChanged WorkerStatusChangedEvent;

        #endregion

        #region work
        private void CollapseEntities()
        {
            Dictionary<int, int> instanceID_CollapsedID_Mappings = new Dictionary<int, int>();
            Dictionary<int, double> instanceID_Score_Mappings = new Dictionary<int, double>();
            SqlConnection conn = new SqlConnection(this._Instruction.ConnStr);
            try
            {
                DataTable dtEntities = null;
                if(this.TestMode)
                {
                    dtEntities = this.GetSampleEntityData();
                    // dtEntities = this.GetEntityData();
                }
                else
                {
                    dtEntities = this.GetEntityData();
                }
                List<string> descFieldNames = new List<string>();
                Dictionary<string, SqlDbType> fieldDbTypes = new Dictionary<string, SqlDbType>();
                Dictionary<string, int> fieldSizes=new Dictionary<string, int>();
                TableColumn[] srcCols = DbUtil.GetTableColumns(
                    DataProviderType.MSSQL, this._Instruction.ConnStr,
                    this._Instruction.SchemaName, this._Instruction.TableName);
                foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                {
                    descFieldNames.Add(fieldName);
                    foreach (TableColumn col in srcCols)
                    {
                        if (col.ColumnName.ToLower() == fieldName.ToLower())
                        {
                            fieldDbTypes.Add(fieldName, col.DbType);
                            fieldSizes.Add(fieldName, col.Size);
                        }
                    }
                }
                List<string> assocFieldNames = new List<string>();
                foreach (string fieldName in this._Instruction.AssociativeAttributes.Keys)
                {
                    assocFieldNames.Add(fieldName);
                    foreach (TableColumn col in srcCols)
                    {
                        if (col.ColumnName.ToLower() == fieldName.ToLower())
                        {
                            fieldDbTypes.Add(fieldName, col.DbType);
                            fieldSizes.Add(fieldName, col.Size);
                        }
                    }
                }

                if (this.WorkerStatusChangedEvent != null)
                {
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName,
                                                  "Clustering entities into groups before collapse...", 5);
                }
                Dictionary<int, List<int>> clusterID_RowNumbers_Mappings = null;
                if (this._Instruction.OrFilters != null && this._Instruction.OrFilters.Count > 0)
                {
                    clusterID_RowNumbers_Mappings =
                        this.ClusterEntitiesUsingFields(
                            dtEntities,
                            this._Instruction.OrFilters);
                }
                else
                {
                    clusterID_RowNumbers_Mappings=new Dictionary<int, List<int>>();
                    List<int> clusteredRows = new List<int>();
                    for(int rowNum =0;rowNum<dtEntities.Rows.Count;rowNum++)
                    {
                        clusteredRows.Add(rowNum);
                    }
                    clusterID_RowNumbers_Mappings.Add(1, clusteredRows);
                }
                int idFieldColIndex = 0;
                for (int i = 0; i < dtEntities.Columns.Count; i++)
                {
                    if (dtEntities.Columns[i].ColumnName.ToLower() == this._Instruction.IDFieldName.ToLower())
                    {
                        idFieldColIndex = i;
                        break;
                    }
                }
                

                Dictionary<string, double> fieldWeights = new Dictionary<string, double>();
                Dictionary<string, double> fieldScoresWhenNotMatched = new Dictionary<string, double>();
                foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                {
                    fieldWeights.Add(fieldName, this._Instruction.DescriptiveAttributes[fieldName]);
                    fieldScoresWhenNotMatched.Add(fieldName, 0);
                }
                foreach (string fieldName in this._Instruction.AssociativeAttributes.Keys)
                {
                    fieldWeights.Add(fieldName, this._Instruction.AssociativeAttributes[fieldName]);
                    fieldScoresWhenNotMatched.Add(fieldName, 0);
                }
                string binFolderPath = Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location);
                // initialize comparers
                Dictionary<string, IAttributeComparer> fieldComparers = new Dictionary<string, IAttributeComparer>();
                foreach (string fieldName in this._Instruction.FieldComparers.Keys)
                {
                    string assemFilePath = Path.Combine(binFolderPath,
                                                        this._Instruction.FieldComparers[fieldName].AssemblyFileName);
                    Assembly assembly = Assembly.LoadFile(assemFilePath);
                    Type comparerType =
                        assembly.GetType(this._Instruction.FieldComparers[fieldName].TypeFullName);
                    IAttributeComparer comparer = (IAttributeComparer)Activator.CreateInstance(comparerType);
                    fieldComparers.Add(fieldName, comparer);
                }
                // initialize compensaters
                List<ICompensater> compensaters = new List<ICompensater>();
                if (this._Instruction.Compensaters != null && this._Instruction.Compensaters.Count > 0)
                {
                    foreach (CompensaterPointer compensaterStruct in this._Instruction.Compensaters)
                    {
                        string assemblyFilePath = Path.Combine(binFolderPath, compensaterStruct.AssemblyName);
                        Assembly assembly = Assembly.LoadFile(assemblyFilePath);
                        Type compensateType = assembly.GetType(compensaterStruct.TypeFullName);
                        ICompensater compensater = (ICompensater)Activator.CreateInstance(compensateType);
                        compensater.CanSwap = compensaterStruct.CanSwap;
                        compensater.CompensateScore = compensaterStruct.CompensateScore;
                        compensater.CompositeFieldNames = compensaterStruct.FieldNames;
                        compensaters.Add(compensater);
                    }
                }

                if (this.WorkerStatusChangedEvent != null)
                {
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName,
                                                  "Collecting field distribution statistics...", 10);
                }
                conn.Open();


                // populate unevenly distributed field value stats
                Dictionary<string, string> fieldValueDistributionTempTableNames =new Dictionary<string, string>();
                if (this._Instruction.UnevenlyDistributedFields != null && 
                    this._Instruction.UnevenlyDistributedFields.Count > 0)
                {
                    fieldValueDistributionTempTableNames =
                        this.PopulateTableFieldDistributionFrequencies(
                            conn, this._Instruction.FieldDistributionSourceTableName,
                            this._Instruction.UnevenlyDistributedFields,
                            fieldDbTypes);
                }

                int entityProcessed = 0;
                int clusterProcessed = 0;
                Dictionary<int,int> collapseIDsToMerge =new Dictionary<int, int>();
                int totalClusteredRowCount = 0;
                foreach (int clusterID in clusterID_RowNumbers_Mappings.Keys)
                {
                    List<int> rowNumbers = clusterID_RowNumbers_Mappings[clusterID];
                    totalClusteredRowCount += rowNumbers.Count;
                }
                foreach (int clusterID in clusterID_RowNumbers_Mappings.Keys)
                {
                    List<int> rowNumbers = clusterID_RowNumbers_Mappings[clusterID];
                    if (this.WorkerStatusChangedEvent != null && clusterProcessed % 50==0)
                    {
                        string msg =
                            string.Format(
                                "{0} processed: {1} of {2}, total collapsed={3}, neighbor clusters={4}. Trying to collapsing {5} entities...",
                                this._Instruction.TableName,
                                entityProcessed,
                                totalClusteredRowCount,
                                this._LastCollapseID,
                                collapseIDsToMerge.Count,
                                rowNumbers.Count);
                        int percent = (entityProcessed * 100 / totalClusteredRowCount) % 100;
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                            this.WorkflowName, this.ComponentName, msg, percent);
                    }
                    this.TryCollapseEntities(conn,
                                             dtEntities,
                                             idFieldColIndex,
                                             descFieldNames,
                                             assocFieldNames,
                                             fieldDbTypes,
                                             fieldWeights,
                                             fieldScoresWhenNotMatched,
                                             fieldValueDistributionTempTableNames,
                                             fieldComparers,
                                             compensaters,
                                             this._Instruction.CollapseThreshold,
                                             rowNumbers,
                                             ref instanceID_CollapsedID_Mappings,
                                             ref instanceID_Score_Mappings, 
                                             ref collapseIDsToMerge);
                    entityProcessed += rowNumbers.Count;
                    clusterProcessed++;
                }

                
                Dictionary<int, List<int>> collapseID_InstanceIDs_Mappings = new Dictionary<int, List<int>>();
                foreach (int instanceID in instanceID_CollapsedID_Mappings.Keys)
                {
                    int collapseID = instanceID_CollapsedID_Mappings[instanceID];
                    if (collapseID_InstanceIDs_Mappings.ContainsKey(collapseID))
                    {
                        List<int> instanceIDs = collapseID_InstanceIDs_Mappings[collapseID];
                        instanceIDs.Add(instanceID);
                        collapseID_InstanceIDs_Mappings[collapseID] = instanceIDs;
                    }
                    else
                    {
                        List<int> instanceIDs = new List<int>();
                        instanceIDs.Add(instanceID);
                        collapseID_InstanceIDs_Mappings.Add(collapseID, instanceIDs);
                    }
                }

                // merge neighbor collapses, 
                if (collapseIDsToMerge.Count > 0)
                {
                    if(this.WorkerStatusChangedEvent !=null)
                    {
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName,
                                                      "Trying to merge neighbor collapses...", 95);
                    }

                    List<int> highToLowCollapseIDs=new List<int>();
                    foreach(int neighborCollapseID in collapseIDsToMerge.Keys)
                    {
                        highToLowCollapseIDs.Add(neighborCollapseID);
                    }
                    highToLowCollapseIDs.Sort();
                    for (int i = highToLowCollapseIDs.Count - 1; i >= 0; i--)
                    {
                        int highCollapseID = highToLowCollapseIDs[i];
                        int lowCollapseID = collapseIDsToMerge[highCollapseID];
                        List<int> instanceIDs = collapseID_InstanceIDs_Mappings[highCollapseID];
                        foreach (int instanceID in instanceIDs)
                        {
                            instanceID_CollapsedID_Mappings[instanceID] = lowCollapseID;
                        }
                    }
                    foreach(int collapseIDToRemove in highToLowCollapseIDs)
                    {
                        collapseID_InstanceIDs_Mappings.Remove(collapseIDToRemove);
                    }
                }

                if (this.WorkerStatusChangedEvent != null)
                {
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName,
                                                  "Saving output to table...", 95);
                }
                this.SaveOutput(this._Instruction.ConnStr,
                                this._Instruction.OutputTableName,
                                instanceID_CollapsedID_Mappings,
                                instanceID_Score_Mappings);
                this.SaveCollapsedResult(
                    this._Instruction.ConnStr,
                    this._Instruction.CollapsedTableName,
                    dtEntities,
                    instanceID_CollapsedID_Mappings,
                    descFieldNames,
                    fieldDbTypes,
                    fieldSizes,
                    fieldComparers,
                    fieldValueDistributionTempTableNames);
                this.UpdateSimilarityScores(
                    instanceID_CollapsedID_Mappings,
                    dtEntities,
                    fieldComparers,
                    fieldValueDistributionTempTableNames,
                    ref instanceID_Score_Mappings);

                this._Output = new EntityCollapseOutput(this._Instruction.OwnerComponentID, this._JobTicketID,
                                                        collapseID_InstanceIDs_Mappings.Count);

                if (this.TestMode)
                {
                    DataTable dtUnderCollapsed =
                            this.GetUnderCollapsedEntities(
                                dtEntities,
                                instanceID_CollapsedID_Mappings,
                                instanceID_Score_Mappings,
                                clusterID_RowNumbers_Mappings,
                                fieldComparers,
                                fieldValueDistributionTempTableNames);
                    if (this.OnUnderCollapsedEntitiesPopulated != null)
                    {
                        string fkName = "CollapsedID";
                        this.OnUnderCollapsedEntitiesPopulated(dtUnderCollapsed, this._Instruction.CollapsedTableName, fkName);
                    }
                    if (!string.IsNullOrEmpty(this._Instruction.UnderCollapsedFilePath))
                    {
                        this.SaveUnderCollapsed(dtUnderCollapsed, this._Instruction.UnderCollapsedFilePath);
                    }

                    DataTable dtOverCollapsed =
                        this.GetOverCollapsedEntities(
                            dtEntities,
                            instanceID_CollapsedID_Mappings,
                            instanceID_Score_Mappings,
                            fieldComparers,
                            conn,
                            fieldValueDistributionTempTableNames);
                    if (this.OnOverCollapsedEntitiesPopulated != null)
                    {
                        string fkName = "CollapsedID";
                        this.OnOverCollapsedEntitiesPopulated(dtOverCollapsed, this._Instruction.CollapsedTableName, fkName);
                    }
                    if (!string.IsNullOrEmpty(this._Instruction.OverCollapsedFilePath))
                    {
                        this.SaveOverCollapsed(dtOverCollapsed, this._Instruction.OverCollapsedFilePath);
                    }
                }

                if(!string.IsNullOrEmpty(this._Instruction.FinalResultFilePath))
                {
                    this.SaveFinalResults(this._Instruction.FinalResultFilePath, CollapseIDFieldName, "SimilarityScore");
                }
                
                if (this.TestMode)
                {
                    Dictionary<string, DescriptiveAttributeContribution> fieldContributions =
                        this.GetDescriptiveAttributeContributions(dtEntities, fieldDbTypes,
                                                                  instanceID_CollapsedID_Mappings);
                    if (this.OnFieldContributionsUpdated != null)
                        this.OnFieldContributionsUpdated(fieldContributions);
                }

                if (this.WorkerFinishedEvent != null)
                {
                    this.WorkerFinishedEvent(this._JobTicketID, this.GetType(), this._Instruction, this._Output);
                }
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                conn.Close();
            }
        }
        #endregion

        #region calculate field distribution
        private Dictionary<string,string> PopulateTableFieldDistributionFrequencies(
            SqlConnection conn, string tblName, 
            List<string> unevenFieldNames, 
            Dictionary<string,SqlDbType> fieldDbTypes)
        {
            ResourceManager resMgr =
                    new ResourceManager(this.GetType().Namespace + ".SQL", Assembly.GetAssembly(this.GetType()));
            string scriptBody = resMgr.GetString("PopulateKeyPersonnelFieldValueDistributionStats");

            //string binFolderPath = Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location);
            //string collapseFolderPath = Path.Combine(binFolderPath, "Collapse");
            //string sqlFilePath = Path.Combine(collapseFolderPath, "PopulateKeyPersonnelFieldValueDistributionStats.sql");
            //if(!File.Exists(sqlFilePath))
            //{
            //    throw new Exception("Unable to find sql script to create field dist population stored proc");
            //}
            //StreamReader fileReader =new StreamReader(sqlFilePath);
            //string scriptBody = fileReader.ReadToEnd();
            //fileReader.Close();
            bool createdSP = DbUtil.NewStoredProc(this._Instruction.ConnStr, PopulateFieldValueDistributionStoredProcName, scriptBody);
            if (!createdSP)
                throw new Exception("Unable to create stored procedure");

            Dictionary<string,string> freqDistTblNames=new Dictionary<string, string>();

            if(conn.State!=ConnectionState.Open)
                conn.Open();

            SqlCommand cmd = conn.CreateCommand();
            cmd.CommandType = CommandType.StoredProcedure;
            cmd.CommandText = PopulateFieldValueDistributionStoredProcName;

            SqlParameter tableNameParam = new SqlParameter("@TableName", tblName);
            SqlParameter fieldNameParam=new SqlParameter("@FieldName",DBNull.Value);
            SqlParameter fieldDbTypeParam=new SqlParameter("@FieldDbType",DBNull.Value);
            SqlParameter distTableNameParam=new SqlParameter("@FieldValueDistributionTableName",DBNull.Value);

            foreach (string fieldName in unevenFieldNames)
            {
                string freqTblName = "tmp_FieldValueDistribution_" + fieldName;
                FieldDataType dataType = new FieldDataType();
                dataType.DataType = fieldDbTypes[fieldName];
                dataType.Nullible = false;
                dataType.Precision = 0;
                dataType.Provider = DataProviderType.MSSQL;
                dataType.Scale = 0;
                dataType.Size = FieldDataType.GetFieldDefaultSize(fieldDbTypes[fieldName]);
                string dbTypeStr = FieldDataType.ToMssqlFieldDataTypeString(dataType);
                fieldNameParam.Value = fieldName;
                fieldDbTypeParam.Value = dbTypeStr;
                distTableNameParam.Value = freqTblName;
                cmd.Parameters.Clear();
                cmd.Parameters.AddRange(new SqlParameter[]
                                            {tableNameParam, fieldNameParam, fieldDbTypeParam, distTableNameParam});
                cmd.ExecuteNonQuery();
                freqDistTblNames.Add(fieldName, freqTblName);
            }

            return freqDistTblNames;
        }
        #endregion

        #region retrieve src data
        private DataTable GetEntityData()
        {
            DataTable dtSrcData = null;
            SqlConnection conn=new SqlConnection(this._Instruction.ConnStr);
            try
            {
                conn.Open();
                SqlCommand cmd = conn.CreateCommand();
                List<string> fields=new List<string>();
                fields.Add(this._Instruction.IDFieldName);
                if (this._Instruction.DescriptiveAttributes != null && this._Instruction.DescriptiveAttributes.Count > 0)
                {
                    foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                    {
                        if (!fields.Contains(fieldName))
                        {
                            fields.Add(fieldName);
                        }
                    }
                }
                if (this._Instruction.AssociativeAttributes != null && this._Instruction.AssociativeAttributes.Count > 0)
                {
                    foreach (string fieldName in this._Instruction.AssociativeAttributes.Keys)
                    {
                        if (!fields.Contains(fieldName))
                        {
                            fields.Add(fieldName);
                        }
                    }
                }

                string selectSql = "select ";
                for(int i=0;i<fields.Count;i++)
                {
                    selectSql += fields[i];
                    if(i<fields.Count -1)
                    {
                        selectSql += ",";
                    }
                }
                selectSql += " from "
                             + (string.IsNullOrEmpty(this._Instruction.SchemaName)
                                    ? ""
                                    : this._Instruction.SchemaName + ".") +
                             this._Instruction.TableName;
                cmd.CommandText = selectSql;
                SqlDataAdapter daSelect=new SqlDataAdapter(cmd);
                DataSet dsSelect=new DataSet();
                daSelect.Fill(dsSelect);
                dtSrcData = dsSelect.Tables[0];
            }
            catch(Exception ex)
            {
                if(this.WorkerErredEvent !=null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                conn.Close();
            }
            return dtSrcData;
        }

        private Dictionary<string,string> GetEntityFieldValues(int entityID)
        {
            DataTable dtSrcData = null;
            SqlConnection conn = new SqlConnection(this._Instruction.ConnStr);
            try
            {
                conn.Open();
                SqlCommand cmd = conn.CreateCommand();
                List<string> fields = new List<string>();
                fields.Add(this._Instruction.IDFieldName);
                if (this._Instruction.DescriptiveAttributes != null && this._Instruction.DescriptiveAttributes.Count > 0)
                {
                    foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                    {
                        if (!fields.Contains(fieldName))
                        {
                            fields.Add(fieldName);
                        }
                    }
                }
                if (this._Instruction.AssociativeAttributes != null && this._Instruction.AssociativeAttributes.Count > 0)
                {
                    foreach (string fieldName in this._Instruction.AssociativeAttributes.Keys)
                    {
                        if (!fields.Contains(fieldName))
                        {
                            fields.Add(fieldName);
                        }
                    }
                }

                string selectSql = "select ";
                for (int i = 0; i < fields.Count; i++)
                {
                    selectSql += fields[i];
                    if (i < fields.Count - 1)
                    {
                        selectSql += ",";
                    }
                }
                selectSql += " from "
                             + (string.IsNullOrEmpty(this._Instruction.SchemaName)
                                    ? ""
                                    : this._Instruction.SchemaName + ".") +
                             this._Instruction.TableName;
                selectSql += " where " + this._Instruction.IDFieldName + "=" + entityID;

                cmd.CommandText = selectSql;
                SqlDataAdapter daSelect = new SqlDataAdapter(cmd);
                DataSet dsSelect = new DataSet();
                daSelect.Fill(dsSelect);
                dtSrcData = dsSelect.Tables[0];
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                conn.Close();
            }

            Debug.Assert(dtSrcData != null && dtSrcData.Rows.Count == 1, "Unable to retrieve data by ID " + entityID);

            Dictionary<string,string> fieldValues=new Dictionary<string, string>();
            for(int i=0;i<dtSrcData.Columns.Count;i++)
            {
                object fieldValue = dtSrcData.Rows[0][i];
                fieldValues.Add(dtSrcData.Columns[i].ColumnName, fieldValue == null ? null : fieldValue.ToString());
            }
            return fieldValues;
        }
        #endregion

        #region pairwise comparison
        private bool CompareString(ref double compareScore, string value1, string value2, int minLenTolerant, double minPercentSubStringMatch)
        {
            if(!string.IsNullOrEmpty(value1) && !string.IsNullOrEmpty(value2))
            {
                value1 = value1.Trim().ToLower();
                value2 = value2.Trim().ToLower();
                if(value1==value2)
                {
                    compareScore = 1.0;
                    return true;
                }
                else
                {
                    if(value1.Length >=minLenTolerant && value2.Length >=minLenTolerant)
                    {
                        int len = Math.Min(value1.Length, value2.Length);
                        int subStringLenthMatched = StringUtil.LongestCommonSubsequence(value1, value2);
                        if ((double)subStringLenthMatched * 100 / len > minPercentSubStringMatch)
                        {
                            compareScore = ((double) subStringLenthMatched)/len;
                            return true;
                        }
                        else
                        {
                            compareScore = 0;
                            return false;
                        }
                    }
                    else
                    {
                        compareScore = 0;
                        return false;
                    }
                }
            }
            else
            {
                compareScore = 0;
                return false;
            }
        }

        private bool CompareDate(ref double compareScore, DateTime var1, DateTime var2, int maxDigitTransposeAllowed, int minContiguousDigitsMatched)
        {
            if(var1==var2)
            {
                compareScore = 1.0;
                return true;
            }

            string dateStr1 = var1.Year.ToString() +
                              (var1.Month >= 10
                                   ?
                                       var1.Month.ToString()
                                   : "0" + var1.Month.ToString()) +
                              (var1.Day >= 10 ? var1.Day.ToString() : "0" + var1.Day.ToString());
            string dateStr2 = var2.Year.ToString() +
                                          (var2.Month >= 10
                                               ?
                                                   var2.Month.ToString()
                                               : "0" + var2.Month.ToString()) +
                                          (var2.Day >= 10 ? var2.Day.ToString() : "0" + var2.Day.ToString());
            int substringLenMatched = StringUtil.LongestCommonSubsequence(dateStr1, dateStr2);
            if (substringLenMatched <= minContiguousDigitsMatched)
            {
                compareScore = 0;
                return false;
            }
            char[] dateChar1 = dateStr1.ToCharArray();
            char[] dateChar2 = dateStr2.ToCharArray();
            int misMatchCount = 0;
            BitArray bits=new BitArray(dateChar2.Length);
            bits.SetAll(true);
            for(int i=0;i<dateChar2.Length;i++)
            {
                if(dateChar1[i]!=dateChar2[i])
                {
                    bits[i] = false;
                    misMatchCount++;
                }
            }
            if(misMatchCount <=maxDigitTransposeAllowed)
            {
                compareScore = (double) misMatchCount/8;
                return true;
            }
            else
            {
                compareScore = 0;
                return false;
            }
        }
        #endregion

        #region desc field prob calculation
        private void CalculateDescriptiveFieldProbabilities(
            Dictionary<string,TableFieldStats<IComparable>> fieldStats, 
            Dictionary<string,List<ModePercentile>> fieldDistributions,
            ref Dictionary<string,Dictionary<string,double>> probByValues, 
            ref Dictionary<string,double> probByField)
        {
            foreach(string fieldName in this._Instruction.DescriptiveAttributes.Keys)
            {
                // int fieldNonNullValueCountAfterCollapse = 1;
                if(fieldStats.ContainsKey(fieldName))
                {
                    TableFieldStats<IComparable> fieldStat = fieldStats[fieldName];
                    //fieldNonNullValueCountAfterCollapse = (int) (fieldStat.NonNullValueCount*this._Instruction.CollapseRatio);
                    //double prob = (double) fieldStat.DistinctCount/fieldNonNullValueCountAfterCollapse;
                    //prob = Math.Min(1.0, prob);
                    double prob = Math.Log10(fieldStat.DistinctCount) / Math.Log10(fieldStat.NonNullValueCount);
                    probByField.Add(fieldName, prob);
                }
                if(fieldDistributions.ContainsKey(fieldName))
                {
                    List<ModePercentile> modePcts = fieldDistributions[fieldName];
                    double averageFrequency = 0;
                    foreach(ModePercentile modePct in modePcts)
                    {
                        averageFrequency += modePct.AverageFrequency;
                    }
                    averageFrequency = averageFrequency/modePcts.Count;
                    Dictionary<string,double> valueDist=new Dictionary<string, double>();
                    foreach(ModePercentile modePct in modePcts)
                    {
                        double percentileWeight = modePct.AverageFrequency/averageFrequency;
                        double prob = 1.0;
                        if(probByField.ContainsKey(fieldName))
                        {
                            prob = probByField[fieldName];
                        }
                        double pctProb = prob - Math.Log10(percentileWeight);
                        // double pctProb = modePct.Count*normalizedFrequency/fieldNonNullValueCountAfterCollapse;
                        // pctProb = Math.Min(1.0, pctProb);
                        foreach (string distinctValue in modePct.DistinctValues)
                        {
                            valueDist.Add(distinctValue, pctProb);
                        }
                    }
                    probByValues.Add(fieldName, valueDist);
                }
            }
        }
        #endregion

        #region assoc field prob calculation
        private void CalculateAssociativeFieldProbabilities(
            Dictionary<string, TableFieldStats<IComparable>> fieldStats,
            Dictionary<string, List<ModePercentile>> fieldDistributions,
            ref Dictionary<string, Dictionary<string, double>> probByValues, 
            ref Dictionary<string, double> probByField)
        {
            foreach (string fieldName in this._Instruction.AssociativeAttributes.Keys)
            {
                // int fieldNonNullValueCountAfterCollapse = 1;
                if (fieldStats.ContainsKey(fieldName))
                {
                    TableFieldStats<IComparable> fieldStat = fieldStats[fieldName];
                    //fieldNonNullValueCountAfterCollapse = (int)(fieldStat.NonNullValueCount * this._Instruction.CollapseRatio);
                    //double prob = (double)fieldStat.DistinctCount / fieldNonNullValueCountAfterCollapse;
                    //probByField.Add(fieldName, prob);
                    double prob = Math.Log10(fieldStat.DistinctCount) / Math.Log10(fieldStat.NonNullValueCount);
                    probByField.Add(fieldName, prob);
                }
                if (fieldDistributions.ContainsKey(fieldName))
                {
                    List<ModePercentile> modePcts = fieldDistributions[fieldName];
                    double averageFrequency = 0;
                    foreach (ModePercentile modePct in modePcts)
                    {
                        averageFrequency += modePct.AverageFrequency;
                    }
                    averageFrequency = averageFrequency / modePcts.Count;
                    Dictionary<string, double> valueDist = new Dictionary<string, double>();
                    foreach (ModePercentile modePct in modePcts)
                    {
                        double percentileWeight = modePct.AverageFrequency / averageFrequency;
                        double prob = 1.0;
                        if (probByField.ContainsKey(fieldName))
                        {
                            prob = probByField[fieldName];
                        }
                        double pctProb = prob - Math.Log10(percentileWeight);
                        
                        // double pctProb = modePct.Count * normalizedFrequency / fieldNonNullValueCountAfterCollapse;
                        foreach (string distinctValue in modePct.DistinctValues)
                        {
                            valueDist.Add(distinctValue, pctProb);
                        }
                    }
                    probByValues.Add(fieldName, valueDist);
                }
            }
        }
        #endregion

        #region cluster tuples by fields
        /// <summary>
        /// 
        /// </summary>
        /// <param name="dtEntities"></param>
        /// <param name="orFilters">i.e. [["FirstName","LastName"]["LastName","DOB"]["DOB","SSN"]]</param>
        private  Dictionary<int, List<int>> ClusterEntitiesUsingFields(
            DataTable dtEntities, List<List<string>> orFilters)
        {
            Dictionary<int, List<int>> clusterID_RowNumbers = new Dictionary<int, List<int>>();
            
            int lastClusterID = 0;
            Dictionary<string, List<int>> value_RowNumbers_Mappings = new Dictionary<string, List<int>>();
            Dictionary<int,SqlDbType> colIndex_DbType_Mappings=new Dictionary<int, SqlDbType>();
            Dictionary<string,int> colName_ColIndex_Mappings=new Dictionary<string, int>();
            // int idColIndex = 0;
            for(int k=0;k<dtEntities.Columns.Count;k++)
            {
                //if(dtEntities.Columns[k].ColumnName.ToLower()==this._Instruction.IDFieldName.ToLower())
                //{
                //    idColIndex = k;
                //}
                SqlDbType dbType = FieldDataType.ToDbType(dtEntities.Columns[k].DataType.ToString());
                colIndex_DbType_Mappings.Add(k,dbType);
                colName_ColIndex_Mappings.Add(dtEntities.Columns[k].ColumnName.ToLower(), k);
            }

            //List<int> addedInstanceIDs = new List<int>();
            //for(int i=0;i<dtEntities.Rows.Count;i++)
            //{
            //    int instanceID = (int) dtEntities.Rows[i][idColIndex];
            //    if (!addedInstanceIDs.Contains(instanceID))
            //    {
            //        string filterStr = string.Empty;
            //        for (int k = 0; k < orFilters.Count; k++)
            //        {
            //            List<string> andFilters = orFilters[k];
            //            string andFilterStr = "(";
            //            for (int m = 0; m < andFilters.Count; m++)
            //            {
            //                if (dtEntities.Rows[i][andFilters[m]] != null)
            //                {
            //                    andFilterStr +=
            //                        andFilters[m] + "=" +
            //                        FieldDataType.ToMssqlDataValue(
            //                            fieldDbTypes[andFilters[m]],
            //                            dtEntities.Rows[i][andFilters[m]].ToString());
            //                }
            //                else
            //                {
            //                    andFilterStr += andFilters[m] + " is null";
            //                }
            //                if (m < andFilters.Count - 1)
            //                {
            //                    andFilterStr += " and ";
            //                }
            //            }
            //            andFilterStr += ")";
            //            filterStr += andFilterStr;
            //            if (k < orFilters.Count - 1)
            //            {
            //                filterStr += " or ";
            //            }
            //        }

            //        DataRow[] drs = dtEntities.Select(filterStr);
            //        lastClusterID++;
            //        if (drs != null && drs.Length > 0)
            //        {
            //            foreach(DataRow dr in drs)
            //            {
            //                int instanceID2 =(int) dr[idColIndex];
            //                if(!addedInstanceIDs.Contains(instanceID2))
            //                {
            //                    addedInstanceIDs.Add(instanceID2);
            //                }
            //            }
            //            clusterID_RowNumbers.Add(lastClusterID, drs);
            //        }
            //        else
            //        {
            //            clusterID_RowNumbers.Add(lastClusterID, new DataRow[] {dtEntities.Rows[i]});
            //            if (!addedInstanceIDs.Contains(instanceID))
            //            {
            //                addedInstanceIDs.Add(instanceID);
            //            }
            //        }
            //    }
            //}
            //return clusterID_RowNumbers;
            Dictionary<int, int> rowNum_ClusterID_Mappings = new Dictionary<int, int>();
            for (int k = 0; k < orFilters.Count; k++)
            {
                List<string> andFilters = orFilters[k];
                for (int i = 0; i < dtEntities.Rows.Count; i++)
                {
                    string value = string.Empty;
                    bool hasNullField = false;
                    foreach (string filterFieldName in andFilters)
                    {
                        int colIdx = colName_ColIndex_Mappings[filterFieldName.ToLower()];
                        if (dtEntities.Rows[i][colIdx] != null &&
                            dtEntities.Rows[i][colIdx] != DBNull.Value)
                        {
                            if (value.Length > 0)
                            {
                                value += "|";
                            }
                            
                            value += dtEntities.Rows[i][colIdx].ToString().ToLower();
                        }
                        else
                            hasNullField = true;
                    }
                    if (!string.IsNullOrEmpty(value) && (!hasNullField))
                    {
                        if (!value_RowNumbers_Mappings.ContainsKey(value))
                        {
                            List<int> rowNumbers = new List<int>();
                            rowNumbers.Add(i);
                            value_RowNumbers_Mappings.Add(value, rowNumbers);
                        }
                        else
                        {
                            List<int> rowNumbers = value_RowNumbers_Mappings[value];
                            rowNumbers.Add(i);
                            value_RowNumbers_Mappings[value] = rowNumbers;
                        }
                    }

                    if (this.WorkerStatusChangedEvent != null && i % 100 == 0)
                    {
                        string msg = string.Format(
                            "Cluster entities based on unique values from {0}, processed {1} of {2}...",
                            StringUtil.ToString(orFilters[k].ToArray()), i + 1, dtEntities.Rows.Count);
                        int percent = i*100/dtEntities.Rows.Count;
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction,
                            this.WorkflowName, this.ComponentName, msg, percent);
                    }
                }

                

                if (value_RowNumbers_Mappings.Count > 0)
                {
                    int distinctValueProcessed = 0;
                    foreach (string distinctValue in value_RowNumbers_Mappings.Keys)
                    {
                        distinctValueProcessed++;
                        List<int> groupedRowIdxes = value_RowNumbers_Mappings[distinctValue];
                        int currentClusterID = 0;
                        foreach(int rowNum in groupedRowIdxes)
                        {
                            if(rowNum_ClusterID_Mappings.ContainsKey(rowNum))
                            {
                                currentClusterID = rowNum_ClusterID_Mappings[rowNum];
                                break;
                            }
                        }
                        if (currentClusterID > 0)
                        {
                            List<int> existingRowIndexes = clusterID_RowNumbers[currentClusterID];
                            foreach (int rowNum in groupedRowIdxes)
                            {
                                if (!existingRowIndexes.Contains(rowNum))
                                {
                                    existingRowIndexes.Add(rowNum);
                                }
                            }
                            clusterID_RowNumbers[currentClusterID] = existingRowIndexes;
                            foreach (int rowNum in existingRowIndexes)
                            {
                                if (!rowNum_ClusterID_Mappings.ContainsKey(rowNum))
                                {
                                    rowNum_ClusterID_Mappings.Add(rowNum, currentClusterID);
                                }
                            }
                        }
                        else
                        {
                            lastClusterID++;
                            currentClusterID = lastClusterID;
                            clusterID_RowNumbers.Add(currentClusterID, groupedRowIdxes);
                            foreach (int rowNum in groupedRowIdxes)
                            {
                                if (!rowNum_ClusterID_Mappings.ContainsKey(rowNum))
                                {
                                    rowNum_ClusterID_Mappings.Add(rowNum, currentClusterID);
                                }
                            }
                        }

                        if (this.WorkerStatusChangedEvent != null && distinctValueProcessed % 100 == 0)
                        {
                            string msg =
                                string.Format(
                                    "Cluster entities based on unique values from {0}, processed {1} of {2}...total clusters={3}",
                                    StringUtil.ToString(orFilters[k].ToArray()),
                                    distinctValueProcessed,
                                    value_RowNumbers_Mappings.Count,
                                    clusterID_RowNumbers.Count);
                            int percent = distinctValueProcessed*100/value_RowNumbers_Mappings.Count;
                            this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName, msg,
                                                          percent);
                        }
                    }
                }

                value_RowNumbers_Mappings.Clear();
            }

            //// populate row# <- list of cluster ids
            //Dictionary<int, List<int>> rowNum_ClusterIDs = new Dictionary<int, List<int>>();
            //foreach (int clusterID in clusterID_RowNumbers.Keys)
            //{
            //    List<int> rowNumbers = clusterID_RowNumbers[clusterID];
            //    foreach (int rowNum in rowNumbers)
            //    {
            //        if (rowNum_ClusterIDs.ContainsKey(rowNum))
            //        {
            //            List<int> clusterIDs = rowNum_ClusterIDs[rowNum];
            //            clusterIDs.Add(clusterID);
            //            rowNum_ClusterIDs[rowNum] = clusterIDs;
            //        }
            //        else
            //        {
            //            List<int> clusterIDs = new List<int>();
            //            clusterIDs.Add(clusterID);
            //            rowNum_ClusterIDs.Add(rowNum, clusterIDs);
            //        }
            //    }
            //}

            //List<int> visitedRowNumbers = new List<int>();
            //List<int> visitedClusterIDs = new List<int>();
            //Dictionary<int, List<int>> groupID_ClusterIDs = new Dictionary<int, List<int>>();
            //int lastGroupID = 0;
            //foreach (int clusterID in clusterID_RowNumbers.Keys)
            //{
            //    if (!visitedClusterIDs.Contains(clusterID))
            //    {
            //        List<int> groupedClusterIDs = new List<int>();
            //        groupedClusterIDs.Add(clusterID);
            //        List<int> groupedRowNums = new List<int>();

            //        this.WalkFromClusterIDToRowNumber(
            //            clusterID, clusterID_RowNumbers, rowNum_ClusterIDs,
            //            ref groupedClusterIDs, ref groupedRowNums);

            //        int currentGroupID = 0;
            //        foreach (int groupID in groupID_ClusterIDs.Keys)
            //        {
            //            List<int> clusterIDsA = groupID_ClusterIDs[groupID];
            //            if (IntUtil.Intercept(clusterIDsA, groupedClusterIDs))
            //            {
            //                currentGroupID = groupID;
            //                break;
            //            }
            //        }
            //        if (currentGroupID == 0)
            //        {
            //            lastGroupID++;
            //            currentGroupID = lastGroupID;
            //            groupID_ClusterIDs.Add(currentGroupID, groupedClusterIDs);
            //        }
            //        else
            //        {
            //            List<int> existingClusterIDs = groupID_ClusterIDs[currentGroupID];
            //            foreach (int groupedClusterID in groupedClusterIDs)
            //            {
            //                if (!existingClusterIDs.Contains(groupedClusterID))
            //                {
            //                    existingClusterIDs.Add(groupedClusterID);
            //                }
            //            }
            //            groupID_ClusterIDs[currentGroupID] = existingClusterIDs;
            //        }

            //        foreach (int visitedClusterID in groupedClusterIDs)
            //        {
            //            if (!visitedClusterIDs.Contains(visitedClusterID))
            //            {
            //                visitedClusterIDs.Add(visitedClusterID);
            //            }
            //        }
            //        foreach (int visitedRowNum in groupedRowNums)
            //        {
            //            if (!visitedRowNumbers.Contains(visitedRowNum))
            //            {
            //                visitedRowNumbers.Add(visitedRowNum);
            //            }
            //        }
            //    }
            //}
            int maxCountWithinGroup = 0;
            foreach(int clusterID in clusterID_RowNumbers.Keys)
            {
                if (clusterID_RowNumbers[clusterID].Count > maxCountWithinGroup)
                    maxCountWithinGroup = clusterID_RowNumbers[clusterID].Count;
            }
            for (int i = 0; i < dtEntities.Rows.Count; i++)
            {
                if (!rowNum_ClusterID_Mappings.ContainsKey(i))
                {
                    lastClusterID++;
                    List<int> rowNumbers = new List<int>();
                    rowNumbers.Add(i);
                    clusterID_RowNumbers.Add(lastClusterID, rowNumbers);
                }
            }
            

            if(this.WorkerStatusChangedEvent !=null)
            {
                string msg = string.Format("Total groups clustered={0}, max number of entities within a group is: {1}",
                                           clusterID_RowNumbers.Count, maxCountWithinGroup);
                int percent = 99;
                this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName, msg, percent);
            }
            return clusterID_RowNumbers;
        }

        private void WalkFromClusterIDToRowNumber(
            int fromClusterID,
            Dictionary<int, List<int>> clusterID_RowNumbers,
            Dictionary<int, List<int>> rowNum_ClusterIDs,
            ref List<int> walkedClusterIDs,
            ref List<int> walkedRowNums)
        {
            List<int> associatedRowNumbers = clusterID_RowNumbers[fromClusterID];
            foreach (int rowNum in associatedRowNumbers)
            {
                if (!walkedRowNums.Contains(rowNum))
                {
                    walkedRowNums.Add(rowNum);
                    this.WalkFromRowNumberToClusterID(
                        rowNum, clusterID_RowNumbers, rowNum_ClusterIDs,
                        ref walkedClusterIDs, ref walkedRowNums);
                }
            }
        }

        private void WalkFromRowNumberToClusterID(
            int fromRowNum,
            Dictionary<int, List<int>> clusterID_RowNumbers,
            Dictionary<int, List<int>> rowNum_ClusterIDs,
            ref List<int> walkedClusterIDs,
            ref List<int> walkedRowNums)
        {
            List<int> associatedClusterIDs = rowNum_ClusterIDs[fromRowNum];
            foreach(int clusterID in associatedClusterIDs)
            {
                if (!walkedClusterIDs.Contains(clusterID))
                {
                    walkedClusterIDs.Add(clusterID);
                    this.WalkFromClusterIDToRowNumber(
                        clusterID, clusterID_RowNumbers, rowNum_ClusterIDs,
                        ref walkedClusterIDs, ref walkedRowNums);
                }
            }
        }
        #endregion

        #region attribute comparison
        private double CompareAttributeForEntity(object val1, object val2, double penalty, 
            IAttributeComparer comparer, SqlConnection conn, 
            string unevenDistTableName)
        {
            return comparer.Compare(val1, val2, penalty, conn, unevenDistTableName);
        }
        #endregion

        #region pair-wise comparison
        private void TryCollapseEntities(
            SqlConnection conn,
            DataTable dtEntities, 
            int idFieldColIndex, 
            List<string> descFieldNames,
            List<string> assocFieldNames,
            Dictionary<string,SqlDbType> fieldDbTypes,
            Dictionary<string,double> fieldWeights,
            Dictionary<string,double> fieldScoreWhenNotMatched,
            Dictionary<string,string> fieldValueDistributionTempTableNames,
            Dictionary<string,IAttributeComparer> fieldComparers,
            List<ICompensater> compensaters,
            double cutoffScore,
            List<int> rowNumbers, 
            ref Dictionary<int,int> instanceID_CollapsedID_Mappings,
            ref Dictionary<int,double> instanceID_Score_Mappings,
            ref Dictionary<int, int> clusterIDToMerge)
        {
            // int iterationCount = 0;
            for (int i = 0; i < rowNumbers.Count; i++)
            {
                DataRow dr1 = dtEntities.Rows[rowNumbers[i]]; // drs[i];
                int instanceID1 = int.Parse(dr1[idFieldColIndex].ToString());
                bool isInstance1AlreadyCollapsed = false;
                if(instanceID_CollapsedID_Mappings.ContainsKey(instanceID1))
                {
                    isInstance1AlreadyCollapsed = true;
                }

                bool collapsed = false;
                Dictionary<string, object> fieldValues1 = new Dictionary<string, object>();
                foreach (string fieldName in descFieldNames)
                {
                    if(dr1[fieldName]!=null && dr1[fieldName]!=DBNull.Value)
                    {
                        fieldValues1.Add(fieldName,dr1[fieldName]);
                    }
                    else
                    {
                        fieldValues1.Add(fieldName,null);
                    }
                }
                foreach (string fieldName in assocFieldNames)
                {
                    if (dr1[fieldName] != null && dr1[fieldName] != DBNull.Value)
                    {
                        fieldValues1.Add(fieldName, dr1[fieldName]);
                    }
                    else
                    {
                        fieldValues1.Add(fieldName, null);
                    }
                }
                for (int k = i + 1; k < rowNumbers.Count; k++)
                {
                    DataRow dr2 = dtEntities.Rows[rowNumbers[k]]; // drs[k];
                    int instanceID2 = int.Parse(dr2[idFieldColIndex].ToString());
                    //iterationCount++;
                    
                    bool isInstance2AlreadyCollapsed = false;
                    if (instanceID_CollapsedID_Mappings.ContainsKey(instanceID2))
                    {
                        isInstance2AlreadyCollapsed = true;
                    }

                    Dictionary<string, object> fieldValues2 = new Dictionary<string, object>();
                    foreach (string fieldName in descFieldNames)
                    {
                        if (dr2[fieldName] != null && dr2[fieldName] != DBNull.Value)
                        {
                            fieldValues2.Add(fieldName, dr2[fieldName]);
                        }
                        else
                        {
                            fieldValues2.Add(fieldName, null);
                        }
                    }
                    foreach (string fieldName in assocFieldNames)
                    {
                        if (dr2[fieldName] != null && dr2[fieldName] != DBNull.Value)
                        {
                            fieldValues2.Add(fieldName, dr2[fieldName]);
                        }
                        else
                        {
                            fieldValues2.Add(fieldName, null);
                        }
                    }
                    
                    // swaps
                    if(this._Instruction.SwappedFieldsCollection !=null && this._Instruction.SwappedFieldsCollection.Count>0)
                    {
                        for(int m=0;m<this._Instruction.SwappedFieldsCollection.Count;m++)
                        {
                            List<string> swappableFields = this._Instruction.SwappedFieldsCollection[m];
                            foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                            {
                                if (swappableFields.Contains(fieldName))
                                {
                                    for(int fromIdx=0;fromIdx <swappableFields.Count-1;fromIdx++)
                                    {
                                        string fromFieldName = swappableFields[fromIdx];
                                        object fromValue1 = fieldValues1[fromFieldName];
                                        object fromValue2 = fieldValues2[fromFieldName];
                                        double fromScore = fieldComparers[fromFieldName].Compare(fromValue1, fromValue2,
                                                                                                 0, null, null);

                                        for(int toIdx=fromIdx +1;toIdx<swappableFields.Count;toIdx++)
                                        {
                                            string toFieldName = swappableFields[toIdx];
                                            object toValue1 = fieldValues1[toFieldName];
                                            object toValue2 = fieldValues2[toFieldName];
                                            double toScore = 
                                                fieldComparers[toFieldName].Compare(toValue1, toValue2, 0, null, null);

                                            if (fromScore==0 && toScore==0)
                                            {
                                                double fromScoreSwapped =
                                                    fieldComparers[fromFieldName].Compare(
                                                    fromValue1, toValue2, 0, null, null);
                                                double toScoreSwapped = fieldComparers[toFieldName].Compare(
                                                    toValue1, fromValue2, 0, null, null);

                                                if (fromScoreSwapped > 0 || toScoreSwapped > 0)
                                                {
                                                    object tmp = fieldValues2[fromFieldName];
                                                    fieldValues2[fromFieldName] = fieldValues2[toFieldName];
                                                    fieldValues2[toFieldName] = tmp;
                                                }
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                    

                    Dictionary<string, double> fieldScores = new Dictionary<string, double>();
                    foreach (string fieldName in descFieldNames)
                    {
                        string distTblName = string.Empty;
                        if (fieldValueDistributionTempTableNames.ContainsKey(fieldName))
                            distTblName = fieldValueDistributionTempTableNames[fieldName];
                        double penalty = this._Instruction.PenaltiesWhenNotMatched[fieldName];
                        double fieldScore = 
                            this.CompareAttributeForEntity(
                            fieldValues1[fieldName],
                            fieldValues2[fieldName],
                            penalty,
                            fieldComparers[fieldName],
                            conn,
                            distTblName);
                        fieldScores.Add(fieldName, fieldScore);
                    }
                    foreach (string fieldName in assocFieldNames)
                    {
                        string distTblName = string.Empty;
                        if (fieldValueDistributionTempTableNames.ContainsKey(fieldName))
                            distTblName = fieldValueDistributionTempTableNames[fieldName];
                        double penalty = this._Instruction.PenaltiesWhenNotMatched[fieldName];
                        double fieldScore = 
                            this.CompareAttributeForEntity(fieldValues1[fieldName],
                            fieldValues2[fieldName],
                            penalty,
                            fieldComparers[fieldName],
                            conn,
                            distTblName);
                        fieldScores.Add(fieldName, fieldScore);
                    }
                    double similarityScore = 0;
                    this.ShouldCollapse(conn, descFieldNames, assocFieldNames, fieldDbTypes,
                                        fieldValues1, fieldValues2, fieldScores, fieldWeights,
                                        fieldScoreWhenNotMatched, fieldValueDistributionTempTableNames,
                                        compensaters, ref similarityScore);
                    if (similarityScore >= cutoffScore)
                    {
                        collapsed = true;
                        int collapseID = 0;
                        if(isInstance1AlreadyCollapsed && isInstance2AlreadyCollapsed)
                        {
                            int lowClusterID = instanceID_CollapsedID_Mappings[instanceID1];
                            int highClusterID = instanceID_CollapsedID_Mappings[instanceID2];

                            if(lowClusterID > highClusterID)
                            {
                                int tmp = lowClusterID;
                                lowClusterID = highClusterID;
                                highClusterID = tmp;
                            }
                            if(!clusterIDToMerge.ContainsKey(highClusterID))
                            {
                                clusterIDToMerge.Add(highClusterID,lowClusterID);
                            }
                        }
                        else if (isInstance2AlreadyCollapsed)
                        {
                            collapseID = instanceID_CollapsedID_Mappings[instanceID2];
                        }
                        else if (isInstance1AlreadyCollapsed)
                        {
                            collapseID = instanceID_CollapsedID_Mappings[instanceID1];
                        }
                        else
                        {
                            this._LastCollapseID++;
                            collapseID = this._LastCollapseID;
                        }

                        if (!instanceID_CollapsedID_Mappings.ContainsKey(instanceID1))
                        {
                            instanceID_CollapsedID_Mappings.Add(instanceID1, collapseID);
                            instanceID_Score_Mappings.Add(instanceID1, similarityScore);
                            isInstance1AlreadyCollapsed = true;
                        }
                        if (!instanceID_CollapsedID_Mappings.ContainsKey(instanceID2))
                        {
                            instanceID_CollapsedID_Mappings.Add(instanceID2, collapseID);
                            instanceID_Score_Mappings.Add(instanceID2, similarityScore);
                        }
                    }
                }
                if(!collapsed && !isInstance1AlreadyCollapsed)
                {
                    this._LastCollapseID++;
                    int collapseID = this._LastCollapseID;
                    if (!instanceID_CollapsedID_Mappings.ContainsKey(instanceID1))
                    {
                        instanceID_CollapsedID_Mappings.Add(instanceID1, collapseID);
                        instanceID_Score_Mappings.Add(instanceID1, 0);
                    }
                }

                //if (this.WorkerStatusChangedEvent != null && iterationCount > 100 && iterationCount % 100 == 0)
                //{
                //    string msg = string.Format(
                //        "Total entities within group: {0} iteration count={1}, collapsed entity count={2}",
                //        rowNumbers.Count, iterationCount, instanceID_CollapsedID_Mappings.Count);
                //    int percent = i*100/rowNumbers.Count;
                //    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, msg, percent);
                //}
            }
        }
        #endregion

        #region entity collapse decision
        private void ShouldCollapse(
            SqlConnection conn,
            List<string> descFieldNames,
            List<string> assocFieldNames, 
            Dictionary<string,SqlDbType> fieldDbTypes,
            Dictionary<string,object> entity1FieldValues,
            Dictionary<string, object> entity2FieldValues, 
            Dictionary<string,double> fieldSimilarityScores, 
            Dictionary<string,double> fieldWeightsWhenMatched,
            Dictionary<string,double> fieldScoresWhenNotMatched,
            Dictionary<string,string> fieldValueDistributionTempTableNames,
            List<ICompensater> compensaters,
            ref double similarityScore)
        {
            similarityScore = 0;
            

            foreach (string fieldName in descFieldNames)
            {
                if (entity1FieldValues[fieldName] != null && entity2FieldValues[fieldName] != null)
                {
                    double fieldScore = fieldSimilarityScores[fieldName];
                    if(fieldScore==0)
                    {
                        fieldScore = fieldScoresWhenNotMatched[fieldName];
                    }
                    else 
                    {
                        double adjustment = 0;
                        if (this._Instruction.UnevenlyDistributedFields!=null && 
                            this._Instruction.UnevenlyDistributedFields.Contains(fieldName) &&  
                            fieldValueDistributionTempTableNames.ContainsKey(fieldName))
                        {
                            adjustment = this.GetFieldMatchScoreAdjustmentByValueDistribution(
                                conn, fieldName, fieldDbTypes[fieldName],
                                fieldValueDistributionTempTableNames[fieldName],
                                entity1FieldValues[fieldName],
                                entity2FieldValues[fieldName]);
                            // adjustment = 0 - Math.Log10(fieldRelativeDistributionFrequencies[fieldName]);
                        }
                        fieldScore *= (1 + adjustment);
                    }
                    similarityScore += fieldScore * fieldWeightsWhenMatched[fieldName];
                }
            }

            if (similarityScore >= this._Instruction.AssociateThreshold)
            {
                foreach (string fieldName in assocFieldNames)
                {
                    double fieldScore = fieldSimilarityScores[fieldName];

                    double adjustment = 0;
                    if (this._Instruction.UnevenlyDistributedFields != null && 
                        this._Instruction.UnevenlyDistributedFields.Contains(fieldName) &&
                        fieldValueDistributionTempTableNames.ContainsKey(fieldName))
                    {
                        adjustment = this.GetFieldMatchScoreAdjustmentByValueDistribution(
                            conn, fieldName, fieldDbTypes[fieldName],
                            fieldValueDistributionTempTableNames[fieldName],
                            entity1FieldValues[fieldName],
                            entity2FieldValues[fieldName]);
                        // adjustment = 0 - Math.Log10(fieldRelativeDistributionFrequencies[fieldName]);
                    }
                    fieldScore *= (1 + adjustment);

                    similarityScore += fieldScore*fieldWeightsWhenMatched[fieldName];
                }
            }

            if (compensaters != null && compensaters.Count > 0)
            {
                foreach (ICompensater compensater in compensaters)
                {
                    similarityScore += compensater.Compensate(fieldSimilarityScores);
                }
            }
        }

        private double GetFieldMatchScoreAdjustmentByValueDistribution(
            SqlConnection conn, 
            string fieldName, 
            SqlDbType dbType, 
            string freqDistTableName, 
            object fieldValue1, 
            object fieldValue2)
        {
            double adjustment1 = 0;
            double adjustment2 = 0;
            SqlCommand cmd = conn.CreateCommand();
            cmd.CommandType = CommandType.Text;
            if (fieldValue1!=null)
            {
                cmd.CommandText = string.Format("select LogRatio from {0} where {1}={2}", freqDistTableName, fieldName,
                                                FieldDataType.ToMssqlDataValue(dbType, fieldValue1.ToString()));
                try
                {
                    adjustment1 = (double) cmd.ExecuteScalar();
                }
                catch
                {
                    adjustment1 = 0;
                }
            }
            if (fieldValue2!=null)
            {
                cmd.CommandText = string.Format("select LogRatio from {0} where {1}={2}", freqDistTableName, fieldName,
                                                FieldDataType.ToMssqlDataValue(dbType, fieldValue2.ToString()));
                try
                {
                    adjustment2 = (double) cmd.ExecuteScalar();
                }
                catch
                {
                    adjustment2 = 0;
                }
            }
            return Math.Min(adjustment1, adjustment2);
        }
        #endregion

        #region test/summary
        ///<summary>
        ///</summary>
        ///<param name="ida"></param>
        ///<param name="idb"></param>
        ///<returns></returns>
        public List<EntityCompareResult> CompareEntity(int ida, int idb)
        {
            List<EntityCompareResult> compareResults = new List<EntityCompareResult>();
            SqlConnection conn = new SqlConnection(this._Instruction.ConnStr);
            try
            {
                conn.Open();
                DataTable dtEntities = this.GetSampleEntitiesToCompare(new int[] {ida, idb});
                DataRow[] dr1 = dtEntities.Select(this._Instruction.IDFieldName + "=" + ida);
                DataRow[] dr2 = dtEntities.Select(this._Instruction.IDFieldName + "=" + idb);
                Debug.Assert(dr1 != null && dr1.Length == 1);
                Debug.Assert(dr2 != null && dr2.Length == 1);

                string binFolderPath = Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location);
                Dictionary<string,IAttributeComparer> fieldComparers=new Dictionary<string, IAttributeComparer>();
                foreach(string fieldName in this._Instruction.FieldComparers.Keys)
                {
                    FieldComparerPointer fieldComparer = this._Instruction.FieldComparers[fieldName];
                    string assemblyFilePath1 = Path.Combine(binFolderPath, fieldComparer.AssemblyFileName);
                    Assembly assembly = Assembly.LoadFile(assemblyFilePath1);
                    Type compareType = assembly.GetType(fieldComparer.TypeFullName);
                    IAttributeComparer comparer = (IAttributeComparer)Activator.CreateInstance(compareType);
                    fieldComparers.Add(fieldName, comparer);
                }

                // swaps
                if (this._Instruction.SwappedFieldsCollection != null && this._Instruction.SwappedFieldsCollection.Count > 0)
                {
                    for (int m = 0; m < this._Instruction.SwappedFieldsCollection.Count; m++)
                    {
                        List<string> swappableFields = this._Instruction.SwappedFieldsCollection[m];
                        foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                        {
                            if (swappableFields.Contains(fieldName))
                            {
                                for (int fromIdx = 0; fromIdx < swappableFields.Count - 1; fromIdx++)
                                {
                                    string fromFieldName = swappableFields[fromIdx];
                                    object fromVar1 = dr1[0][fromFieldName];
                                    object fromVar2 = dr2[0][fromFieldName];
                                    IAttributeComparer fromComparer = fieldComparers[fromFieldName];
                                    double fromScore = fromComparer.Compare(fromVar1, fromVar2, 0, null, null);

                                    for (int toIdx = fromIdx + 1; toIdx < swappableFields.Count; toIdx++)
                                    {
                                        string toFieldName = swappableFields[toIdx];
                                        object toVar1 = dr1[0][toFieldName];
                                        object toVar2 = dr2[0][toFieldName];
                                        IAttributeComparer toComparer = fieldComparers[toFieldName];
                                        double toScore = toComparer.Compare(toVar1, toVar2, 0, null, null);

                                        if (fromScore==0 && toScore==0)
                                        {
                                            double fromScoreSwapped = 
                                                fromComparer.Compare(fromVar1, toVar2, 0, null, null);
                                            //double toScoreSwapped =
                                            //    toComparer.Compare(toVar1, fromVar2, 0, null, null);
                                            if (fromScoreSwapped > fromScore)
                                            {
                                                fromScore = fromScoreSwapped;
                                                object tmp = dr2[0][fromFieldName];
                                                dr2[0][fromFieldName] = dr2[0][toFieldName];
                                                dr2[0][toFieldName] = tmp;
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                }

                foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                {
                    EntityCompareResult result = new EntityCompareResult();
                    result.FieldName = fieldName;
                    result.FieldValueA = dr1[0][fieldName] == null ? string.Empty : dr1[0][fieldName].ToString();
                    result.FieldValueB = dr2[0][fieldName] == null ? string.Empty : dr2[0][fieldName].ToString();
                    result.FieldWeight = this._Instruction.DescriptiveAttributes[fieldName];
                    result.FieldProbability = 1;
                    result.FieldCompareScore = 0;
                    double penalty =  this._Instruction.PenaltiesWhenNotMatched[fieldName];
                    
                    if (!string.IsNullOrEmpty(result.FieldValueA) && !string.IsNullOrEmpty(result.FieldValueB))
                    {
                        IAttributeComparer comparer = fieldComparers[fieldName];
                        string fieldDistTableName = string.Empty;
                        if (this._Instruction.UnevenlyDistributedFields != null && 
                            this._Instruction.UnevenlyDistributedFields.Contains(fieldName))
                        {
                            fieldDistTableName = "tmp_FieldValueDistribution_" + fieldName;
                        }
                        result.FieldCompareScore = comparer.Compare(
                            dr1[0][fieldName], dr2[0][fieldName], penalty,
                            conn, fieldDistTableName);
                        if (this._Instruction.UnevenlyDistributedFields != null && 
                            this._Instruction.UnevenlyDistributedFields.Contains(fieldName))
                        {
                            SqlDbType dbType = SqlDbType.VarChar;
                            foreach (DataColumn col in dtEntities.Columns)
                            {
                                if (col.ColumnName.ToLower() == fieldName.ToLower())
                                {
                                    dbType = FieldDataType.ToDbType(col.DataType.ToString());
                                    break;
                                }
                            }
                            
                            double logRatio =
                                this.GetFieldMatchScoreAdjustmentByValueDistribution(
                                    conn, fieldName, dbType, fieldDistTableName,
                                    result.FieldValueA, result.FieldValueB);
                            result.FieldProbability = Math.Exp(logRatio);
                            result.FieldCompareScore = result.FieldCompareScore*result.FieldProbability;
                        }
                    }
                    result.FieldCompareScore = result.FieldWeight * result.FieldCompareScore;
                    compareResults.Add(result);
                }

                foreach(string fieldName in this._Instruction.AssociativeAttributes.Keys)
                {
                    EntityCompareResult result = new EntityCompareResult();
                    result.FieldName = fieldName;
                    //DataRow[] dr1 = dtEntities.Select(this._Instruction.IDFieldName + "=" + ida);
                    //DataRow[] dr2 = dtEntities.Select(this._Instruction.IDFieldName + "=" + idb);
                    //Debug.Assert(dr1 != null && dr1.Length == 1);
                    //Debug.Assert(dr2 != null && dr2.Length == 1);
                    result.FieldValueA = dr1[0][fieldName] == null ? string.Empty : dr1[0][fieldName].ToString();
                    result.FieldValueB = dr2[0][fieldName] == null ? string.Empty : dr2[0][fieldName].ToString();
                    result.FieldWeight = this._Instruction.AssociativeAttributes[fieldName];
                    result.FieldProbability = 1;
                    result.FieldCompareScore = 0;
                    double penalty = this._Instruction.PenaltiesWhenNotMatched[fieldName];

                    if (!string.IsNullOrEmpty(result.FieldValueA) && !string.IsNullOrEmpty(result.FieldValueB))
                    {
                        IAttributeComparer comparer = fieldComparers[fieldName];
                        string fieldDistTableName = string.Empty;
                        if (this._Instruction.UnevenlyDistributedFields != null && 
                            this._Instruction.UnevenlyDistributedFields.Contains(fieldName))
                        {
                            fieldDistTableName = "tmp_FieldValueDistribution_" + fieldName;
                        }
                        result.FieldCompareScore = comparer.Compare(
                            dr1[0][fieldName], dr2[0][fieldName], penalty,
                            conn, fieldDistTableName);
                        if (this._Instruction.UnevenlyDistributedFields != null && 
                            this._Instruction.UnevenlyDistributedFields.Contains(fieldName))
                        {
                            SqlDbType dbType = SqlDbType.VarChar;
                            foreach (DataColumn col in dtEntities.Columns)
                            {
                                if (col.ColumnName.ToLower() == fieldName.ToLower())
                                {
                                    dbType = FieldDataType.ToDbType(col.DataType.ToString());
                                    break;
                                }
                            }
                            double logRatio =
                                this.GetFieldMatchScoreAdjustmentByValueDistribution(
                                    conn, fieldName, dbType, fieldDistTableName,
                                    result.FieldValueA, result.FieldValueB);
                            result.FieldProbability = Math.Exp(logRatio);
                        }
                    }
                    result.FieldCompareScore = result.FieldWeight * result.FieldCompareScore;
                    compareResults.Add(result);
                }
            }
            catch (Exception ex)
            {
                if(this.WorkerErredEvent !=null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                conn.Close();
            }
            return compareResults;
        }

        private DataTable GetSampleEntitiesToCompare(int[] entityIDs)
        {
            DataTable dtSrcData = null;
            SqlConnection conn = new SqlConnection(this._Instruction.ConnStr);
            try
            {
                conn.Open();
                SqlCommand cmd = conn.CreateCommand();
                List<string> fields = new List<string>();
                fields.Add(this._Instruction.IDFieldName);
                if (this._Instruction.DescriptiveAttributes != null && this._Instruction.DescriptiveAttributes.Count > 0)
                {
                    foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                    {
                        if (!fields.Contains(fieldName))
                        {
                            fields.Add(fieldName);
                        }
                    }
                }
                if (this._Instruction.AssociativeAttributes != null && this._Instruction.AssociativeAttributes.Count > 0)
                {
                    foreach (string fieldName in this._Instruction.AssociativeAttributes.Keys)
                    {
                        if (!fields.Contains(fieldName))
                        {
                            fields.Add(fieldName);
                        }
                    }
                }

                string selectSql = "select ";
                for (int i = 0; i < fields.Count; i++)
                {
                    selectSql += fields[i];
                    if (i < fields.Count - 1)
                    {
                        selectSql += ",";
                    }
                }
                selectSql += " from "
                             + (string.IsNullOrEmpty(this._Instruction.SchemaName)
                                    ? ""
                                    : this._Instruction.SchemaName + ".") +
                             this._Instruction.TableName;
                selectSql += " where " + this._Instruction.IDFieldName + " in (" +
                             IntUtil.ToString(entityIDs) + ")";
                cmd.CommandText = selectSql;
                SqlDataAdapter daSelect = new SqlDataAdapter(cmd);
                DataSet dsSelect = new DataSet();
                daSelect.Fill(dsSelect);
                dtSrcData = dsSelect.Tables[0];
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                conn.Close();
            }
            return dtSrcData;
        }

        private DataTable GetSampleEntityData()
        {
            DataTable dtSrcData = null;
            SqlConnection conn = new SqlConnection(this._Instruction.ConnStr);
            try
            {
                conn.Open();
                SqlCommand cmd = conn.CreateCommand();
                List<string> fields = new List<string>();
                fields.Add(this._Instruction.IDFieldName);
                if (this._Instruction.DescriptiveAttributes != null && this._Instruction.DescriptiveAttributes.Count > 0)
                {
                    foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                    {
                        if (!fields.Contains(fieldName))
                        {
                            fields.Add(fieldName);
                        }
                    }
                }
                if (this._Instruction.AssociativeAttributes != null && this._Instruction.AssociativeAttributes.Count > 0)
                {
                    foreach (string fieldName in this._Instruction.AssociativeAttributes.Keys)
                    {
                        if (!fields.Contains(fieldName))
                        {
                            fields.Add(fieldName);
                        }
                    }
                }

                string selectSql = "select ";
                for (int i = 0; i < fields.Count; i++)
                {
                    selectSql += fields[i];
                    if (i < fields.Count - 1)
                    {
                        selectSql += ",";
                    }
                }
                selectSql += " from "
                             + (string.IsNullOrEmpty(this._Instruction.SchemaName)
                                    ? ""
                                    : this._Instruction.SchemaName + ".") +
                             this._Instruction.TableName;
                selectSql += " tablesample ("+TableSampleSize +" rows)";
                cmd.CommandText = selectSql;
                SqlDataAdapter daSelect = new SqlDataAdapter(cmd);
                DataSet dsSelect = new DataSet();
                daSelect.Fill(dsSelect);
                dtSrcData = dsSelect.Tables[0];
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                conn.Close();
            }
            return dtSrcData;
        }

        private Dictionary<string,DescriptiveAttributeContribution> GetDescriptiveAttributeContributions(
            DataTable dtEntities, 
            Dictionary<string, SqlDbType> fieldDbTypes, 
            Dictionary<int,int> instanceID_CollapseID_Mappings)
        {
            if(this.WorkerStatusChangedEvent !=null)
            {
                this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName,
                                              "Populate field contribution statistics", 5);
            }

            Dictionary<int,List<int>> collapseID_InstanceIDs_Mappings=new Dictionary<int, List<int>>();
            foreach(int instanceID in instanceID_CollapseID_Mappings.Keys)
            {
                int collapseID = instanceID_CollapseID_Mappings[instanceID];
                if(collapseID_InstanceIDs_Mappings.ContainsKey(collapseID))
                {
                    List<int> instanceIDs = collapseID_InstanceIDs_Mappings[collapseID];
                    instanceIDs.Add(instanceID);
                    collapseID_InstanceIDs_Mappings[collapseID] = instanceIDs;
                }
                else
                {
                    List<int> instanceIDs=new List<int>();
                    instanceIDs.Add(instanceID);
                    collapseID_InstanceIDs_Mappings.Add(collapseID,instanceIDs);
                }
            }
            Dictionary<string,DescriptiveAttributeContribution> descAttrContributions=new Dictionary<string, DescriptiveAttributeContribution>();
            int idFieldColIndex = 0;
            for(int i=0;i<dtEntities.Columns.Count;i++)
            {
                if(dtEntities.Columns[i].ColumnName.ToLower()==this._Instruction.IDFieldName.ToLower())
                {
                    idFieldColIndex = i;
                    break;
                }
            }
            List<int> examinedCollapseIDs=new List<int>();
            for(int i=0;i<dtEntities.Rows.Count;i++)
            {
                int instanceID =int.Parse(dtEntities.Rows[i][idFieldColIndex].ToString());
                int collapseID = instanceID_CollapseID_Mappings[instanceID];
                if (!examinedCollapseIDs.Contains(collapseID))
                {
                    bool collapsed = false;
                    if (collapseID_InstanceIDs_Mappings.ContainsKey(collapseID) &&
                        collapseID_InstanceIDs_Mappings[collapseID].Count > 1)
                    {
                        collapsed = true;
                    }
                    if (collapsed)
                    {
                        List<int> instanceIDs = collapseID_InstanceIDs_Mappings[collapseID];
                        // collapsed entities
                        foreach(int instanceID2 in instanceIDs)
                        {
                            DataRow[] drInstance = dtEntities.Select(this._Instruction.IDFieldName + "=" + instanceID2);
                            Debug.Assert(drInstance != null && drInstance.Length == 1,
                                         "Unable to select instance record from table");
                            foreach(string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                            {
                                if(drInstance[0][fieldName]!=null)
                                {
                                    DataRow[] drWithSameFieldValue =
                                    dtEntities.Select(string.Format("{0}={1}",
                                                                    fieldName,
                                                                    FieldDataType.ToMssqlDataValue(
                                                                        fieldDbTypes[fieldName],
                                                                        drInstance[0][fieldName].ToString())));
                                    if (descAttrContributions.ContainsKey(fieldName))
                                    {
                                        DescriptiveAttributeContribution attributeContribution =
                                            descAttrContributions[fieldName];
                                        if (drWithSameFieldValue != null && drWithSameFieldValue.Length > 1)
                                        {
                                            attributeContribution.NumberMatchedInCollapsedEntities += 1;
                                        }
                                        else
                                        {
                                            attributeContribution.NumberNotMatchedInCollapsedEntities += 1;
                                        }
                                        descAttrContributions[fieldName] = attributeContribution;
                                    }
                                    else
                                    {
                                        DescriptiveAttributeContribution attributeContribution =
                                            new DescriptiveAttributeContribution();
                                        attributeContribution.FieldName = fieldName;
                                        attributeContribution.NumberMatchedInNonCollapsedEntities = 0;
                                        attributeContribution.NumberNotMatchedInNonCollapsedEntities = 0;
                                        if (drWithSameFieldValue != null && drWithSameFieldValue.Length > 1)
                                        {
                                            attributeContribution.NumberMatchedInCollapsedEntities = 1;
                                        }
                                        else
                                        {
                                            attributeContribution.NumberNotMatchedInCollapsedEntities = 1;
                                        }
                                        descAttrContributions.Add(fieldName, attributeContribution);
                                    }
                                }
                            }
                        }
                    }
                    else
                    {
                        // non-collapsed entities
                        foreach(string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                        {
                            if (dtEntities.Rows[i][fieldName] != null)
                            {
                                DataRow[] drWithSameFieldValue =
                                    dtEntities.Select(string.Format("{0}={1}",
                                                                    fieldName,
                                                                    FieldDataType.ToMssqlDataValue(
                                                                        fieldDbTypes[fieldName],
                                                                        dtEntities.Rows[i][fieldName].ToString())));
                                if (descAttrContributions.ContainsKey(fieldName))
                                {
                                    DescriptiveAttributeContribution attributeContribution =
                                        descAttrContributions[fieldName];
                                    if(drWithSameFieldValue !=null && drWithSameFieldValue.Length>1)
                                    {
                                        attributeContribution.NumberMatchedInNonCollapsedEntities += 1;
                                    }
                                    else
                                    {
                                        attributeContribution.NumberNotMatchedInNonCollapsedEntities += 1;
                                    }
                                    descAttrContributions[fieldName] = attributeContribution;
                                }
                                else
                                {
                                    DescriptiveAttributeContribution attributeContribution =
                                        new DescriptiveAttributeContribution();
                                    attributeContribution.FieldName = fieldName;
                                    attributeContribution.NumberMatchedInCollapsedEntities = 0;
                                    attributeContribution.NumberNotMatchedInCollapsedEntities = 0;
                                    if(drWithSameFieldValue !=null && drWithSameFieldValue.Length>1)
                                    {
                                        attributeContribution.NumberMatchedInNonCollapsedEntities = 1;
                                    }
                                    else
                                    {
                                        attributeContribution.NumberNotMatchedInNonCollapsedEntities = 1;
                                    }
                                    descAttrContributions.Add(fieldName, attributeContribution);
                                }
                            }
                        }
                    }

                    examinedCollapseIDs.Add(collapseID);
                }

                if(this.WorkerStatusChangedEvent !=null)
                {
                    string msg = string.Format("Collecting field contribution evidence... {0} of {1}", (i + 1),
                                               dtEntities.Rows.Count);
                    int percent = i * 100 / dtEntities.Rows.Count;
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName, msg, percent);
                }
            }
            return descAttrContributions;
        }

        private DataTable GetUnderCollapsedEntities(
            DataTable dtEntities,
            Dictionary<int, int> instanceID_CollapseID_Mappings,
            Dictionary<int, double> instanceID_Score_Mappings,
            Dictionary<int,List<int>> groupID_RowNumbers, 
            Dictionary<string, IAttributeComparer> fieldComparers, 
            Dictionary<string, string> fieldDistTableNames)
        {
            if (this.WorkerStatusChangedEvent != null)
            {
                this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName,
                                              "determining under collapsed entities...", 5);
            }

            Dictionary<int, List<int>> collapseID_InstanceIDs_Mappings = new Dictionary<int, List<int>>();
            foreach (int instanceID in instanceID_CollapseID_Mappings.Keys)
            {
                int collapseID = instanceID_CollapseID_Mappings[instanceID];
                if (collapseID_InstanceIDs_Mappings.ContainsKey(collapseID))
                {
                    List<int> instanceIDs = collapseID_InstanceIDs_Mappings[collapseID];
                    instanceIDs.Add(instanceID);
                    collapseID_InstanceIDs_Mappings[collapseID] = instanceIDs;
                }
                else
                {
                    List<int> instanceIDs = new List<int>();
                    instanceIDs.Add(instanceID);
                    collapseID_InstanceIDs_Mappings.Add(collapseID, instanceIDs);
                }
            }
            int idFieldColIndex = 0;
            for (int i = 0; i < dtEntities.Columns.Count; i++)
            {
                if (dtEntities.Columns[i].ColumnName.ToLower() == this._Instruction.IDFieldName.ToLower())
                {
                    idFieldColIndex = i;
                    break;
                }
            }

            string[] descFieldNames=new string[this._Instruction.DescriptiveAttributes.Count];
            this._Instruction.DescriptiveAttributes.Keys.CopyTo(descFieldNames,0);
            DataTable dtOutput = dtEntities.Clone();
            dtOutput.Columns.Add("CollapsedID", typeof (int));
            dtOutput.Columns.Add("SimilarityScore", typeof (double));
            for(int i=0;i<dtEntities.Columns.Count;i++)
            {
                dtOutput.Columns.Add(
                    dtEntities.Columns[i].ColumnName + "_Compare", 
                    dtEntities.Columns[i].DataType);
            }
            dtOutput.Columns.Add("CollapsedID_Compare", typeof(int));

            int[] groupIDs=new int[groupID_RowNumbers.Count];
            groupID_RowNumbers.Keys.CopyTo(groupIDs, 0);
            int underCollapsedEntityCount = 0;

            for(int i=0;i<groupIDs.Length;i++)
            {
                List<int> neighborRowNumbers = groupID_RowNumbers[groupIDs[i]];
                List<int> collapsedRowNumbers=new List<int>();
                foreach(int rowNum in neighborRowNumbers)
                {
                    int instanceID = int.Parse(dtEntities.Rows[rowNum][idFieldColIndex].ToString());
                    int collapseID = instanceID_CollapseID_Mappings[instanceID];
                    List<int> neighborInstanceIDs = collapseID_InstanceIDs_Mappings[collapseID];
                    if(neighborInstanceIDs.Count>1)
                    {
                        collapsedRowNumbers.Add(rowNum);
                    }
                }
                if(collapsedRowNumbers.Count>0)
                {
                    foreach(int collapsedRowNum in collapsedRowNumbers)
                    {
                        neighborRowNumbers.Remove(collapsedRowNum);
                    }
                }
                if(neighborRowNumbers.Count>0)
                {
                    foreach (int nonCollapsedRowNum in neighborRowNumbers)
                    {
                        int nonCollapsedInstanceID = int.Parse(dtEntities.Rows[nonCollapsedRowNum][idFieldColIndex].ToString());
                        int nonCollapsedCollapseID = instanceID_CollapseID_Mappings[nonCollapsedInstanceID];
                        foreach (int collapsedRowNum in collapsedRowNumbers)
                        {
                            int fieldValueMatchedCount = 0;
                            foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                            {
                                string swappedFieldName = string.Empty;
                                double fieldScore =
                                    this.GetFieldScoreConsideringSwaps(
                                        ref swappedFieldName, null, fieldName,
                                        dtEntities.Rows[nonCollapsedRowNum],
                                        dtEntities.Rows[collapsedRowNum],
                                        fieldComparers, fieldDistTableNames);
                                if (fieldScore > 0)
                                    fieldValueMatchedCount++;
                            }

                            if (fieldValueMatchedCount > 2)
                            {
                                underCollapsedEntityCount++;
                                List<object> itemArray = new List<object>();
                                itemArray.AddRange(dtEntities.Rows[nonCollapsedRowNum].ItemArray);
                                itemArray.Add(nonCollapsedCollapseID);
                                itemArray.Add(instanceID_Score_Mappings[nonCollapsedInstanceID]);
                                itemArray.AddRange(dtEntities.Rows[collapsedRowNum].ItemArray);
                                int entityIDToCompare = int.Parse(dtEntities.Rows[collapsedRowNum][idFieldColIndex].ToString());
                                int collapseIDToCompare = instanceID_CollapseID_Mappings[entityIDToCompare];
                                itemArray.Add(collapseIDToCompare);
                                dtOutput.Rows.Add(itemArray.ToArray());
                            }
                        }
                    }
                }

                if (this.WorkerStatusChangedEvent != null && (i + 1) % 10 == 0)
                {
                    string msg =
                        string.Format(
                            "Determining under-collapsed entities, found {0}, processed {1} of {2} clusters...",
                            underCollapsedEntityCount, (i + 1), groupIDs.Length);
                    int percent = (i + 1)*100/groupIDs.Length;
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName, msg, percent);
                }
            }

            TableHelper.OnSaveTableDataStatusChanged +=
                new TableHelper.SaveTableDataStatusChanged(TableHelper_OnSaveTableDataStatusChanged);
            TableHelper.OnSaveTableDataErred += new TableHelper.SaveTableDataErred(TableHelper_OnSaveTableDataErred);
            TableHelper.SaveTableToDb(this._Instruction.ConnStr, "UnderCollapsedEntities", dtOutput, false);
            TableHelper.OnSaveTableDataStatusChanged -= TableHelper_OnSaveTableDataStatusChanged;
            TableHelper.OnSaveTableDataErred -= TableHelper_OnSaveTableDataErred;
            return dtOutput;
        }

        void TableHelper_OnSaveTableDataErred(string error)
        {
            if (this.WorkerErredEvent != null)
                this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, error,true);
        }

        void TableHelper_OnSaveTableDataStatusChanged(string msg, int percent)
        {
            if (this.WorkerStatusChangedEvent != null)
                this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName, msg, percent);
        }

        private DataTable GetOverCollapsedEntities(
            DataTable dtEntities,
            Dictionary<int, int> instanceID_CollapseID_Mappings,
            Dictionary<int, double> instanceID_Score_Mappings,
            Dictionary<string,IAttributeComparer> fieldComparers,
            SqlConnection conn, 
            Dictionary<string,string> fieldDistTableNames)
        {
            if (this.WorkerStatusChangedEvent != null)
            {
                this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName,
                                              "determining over collapsed entities...", 5);
            }


            Dictionary<int, List<int>> collapseID_InstanceIDs_Mappings = new Dictionary<int, List<int>>();
            foreach (int instanceID in instanceID_CollapseID_Mappings.Keys)
            {
                int collapseID = instanceID_CollapseID_Mappings[instanceID];
                if (collapseID_InstanceIDs_Mappings.ContainsKey(collapseID))
                {
                    List<int> instanceIDs = collapseID_InstanceIDs_Mappings[collapseID];
                    instanceIDs.Add(instanceID);
                    collapseID_InstanceIDs_Mappings[collapseID] = instanceIDs;
                }
                else
                {
                    List<int> instanceIDs = new List<int>();
                    instanceIDs.Add(instanceID);
                    collapseID_InstanceIDs_Mappings.Add(collapseID, instanceIDs);
                }
            }
            int idFieldColIndex = 0;
            for (int i = 0; i < dtEntities.Columns.Count; i++)
            {
                if (dtEntities.Columns[i].ColumnName.ToLower() == this._Instruction.IDFieldName.ToLower())
                {
                    idFieldColIndex = i;
                    break;
                }
            }
            Dictionary<int,int> instanceID_RowNum=new Dictionary<int, int>();
            for(int i=0;i<dtEntities.Rows.Count;i++)
            {
                int instanceID = int.Parse( dtEntities.Rows[i][idFieldColIndex].ToString());
                instanceID_RowNum.Add(instanceID, i);
            }

            DataTable dtOutput = dtEntities.Clone();
            dtOutput.Columns.Add("CollapsedID", typeof (int));
            dtOutput.Columns.Add("SimilarityScore", typeof (double));
            for (int i = 0; i < dtEntities.Columns.Count; i++)
            {
                dtOutput.Columns.Add(dtEntities.Columns[i].ColumnName + "_Compare", dtEntities.Columns[i].DataType);
            }
            dtOutput.Columns.Add("CollapsedID_Compare", typeof (int));

            List<int> examinedCollapseIDs = new List<int>();
            for (int i = 0; i < dtEntities.Rows.Count; i++)
            {
                int instanceID = int.Parse(dtEntities.Rows[i][idFieldColIndex].ToString());
                int collapseID = instanceID_CollapseID_Mappings[instanceID];
                if (!examinedCollapseIDs.Contains(collapseID))
                {
                    List<int> neighborInstanceIDs = collapseID_InstanceIDs_Mappings[collapseID];
                    if (neighborInstanceIDs.Count > 1)
                    {
                        bool isOverCollapsed = false;
                        List<int> overCollapsedRowNumbers=new List<int>();

                        foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                        {
                            string fieldDistTblName = string.Empty;
                            if (fieldDistTableNames.ContainsKey(fieldName))
                                fieldDistTblName = fieldDistTableNames[fieldName];

                            double penalty = this._Instruction.PenaltiesWhenNotMatched[fieldName];
                            List<object> distinctValues = new List<object>();
                            foreach (int neighborID in neighborInstanceIDs)
                            {
                                DataRow[] dr = dtEntities.Select(this._Instruction.IDFieldName + "=" + neighborID);
                                Debug.Assert(dr != null && dr.Length == 1, "Unable to select entity by id");
                                if (dr[0][fieldName] != null && dr[0][fieldName]!=DBNull.Value)
                                {
                                    object value = dr[0][fieldName];
                                    if (!string.IsNullOrEmpty(value.ToString().Trim()))
                                    {
                                        if (!distinctValues.Contains(value))
                                        {
                                            distinctValues.Add(value);
                                        }
                                    }
                                }
                            }
                            if (distinctValues.Count > 1)
                            {
                                IAttributeComparer fieldComparer = fieldComparers[fieldName];
                                object mergedValue = fieldComparer.Merge(
                                    distinctValues, conn, fieldDistTblName);

                                foreach(int neighborInstanceID in neighborInstanceIDs)
                                {
                                    int neighborRowNum = instanceID_RowNum[neighborInstanceID];
                                    object fieldValue = dtEntities.Rows[neighborRowNum][fieldName];
                                    if(fieldValue !=null && fieldValue.ToString().Length>0)
                                    {
                                        double fieldScore = fieldComparer.Compare(
                                            mergedValue, fieldValue, penalty,
                                            conn, fieldDistTblName);
                                        if(fieldScore <=0)
                                        {
                                            isOverCollapsed = true;
                                            if(!overCollapsedRowNumbers.Contains(neighborRowNum))
                                            {
                                                overCollapsedRowNumbers.Add(neighborRowNum);    
                                            }
                                        }
                                    }
                                }
                                
                            }
                        }
                        if (isOverCollapsed && overCollapsedRowNumbers.Count>0)
                        {
                            int rowNumberToCompar = -1;
                            foreach(int neighborInstanceID in neighborInstanceIDs)
                            {
                                int neighborRowNum = instanceID_RowNum[neighborInstanceID];
                                if(!overCollapsedRowNumbers.Contains(neighborRowNum))
                                {
                                    rowNumberToCompar = neighborRowNum;
                                    break;
                                }
                            }
                            if(rowNumberToCompar >=0)
                            {
                                foreach (int neighborRowNum in overCollapsedRowNumbers)
                                {
                                    List<object> itemArray = new List<object>();
                                    itemArray.AddRange(dtEntities.Rows[neighborRowNum].ItemArray);
                                    itemArray.Add(collapseID);
                                    int neighborInstanceID = int.Parse(dtEntities.Rows[neighborRowNum][idFieldColIndex].ToString());
                                    itemArray.Add(instanceID_Score_Mappings[neighborInstanceID]);
                                    itemArray.AddRange(dtEntities.Rows[rowNumberToCompar].ItemArray);
                                    itemArray.Add(collapseID);
                                    dtOutput.Rows.Add(itemArray.ToArray());
                                }
                            }
                        }
                    }
                    
                    examinedCollapseIDs.Add(collapseID);
                }

                if (this.WorkerStatusChangedEvent != null && (i + 1) % 10 == 0)
                {
                    string msg = string.Format("Determining over-collapsed entities... {0} of {1}", (i + 1),
                                               dtEntities.Rows.Count);
                    int percent = i*100/dtEntities.Rows.Count;
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName, msg, percent);
                }
            }


            TableHelper.OnSaveTableDataStatusChanged +=
                new TableHelper.SaveTableDataStatusChanged(TableHelper_OnSaveTableDataStatusChanged);
            TableHelper.OnSaveTableDataErred += new TableHelper.SaveTableDataErred(TableHelper_OnSaveTableDataErred);
            TableHelper.SaveTableToDb(this._Instruction.ConnStr, "OverCollapsedEntities", dtOutput, false);
            TableHelper.OnSaveTableDataStatusChanged -= TableHelper_OnSaveTableDataStatusChanged;
            TableHelper.OnSaveTableDataErred -= TableHelper_OnSaveTableDataErred;
            return dtOutput;
        }

        #endregion

        #region save output
        private void SaveOutput(
            string connStr, 
            string outputTableName, 
            Dictionary<int,int> instanceID_CollapseID_Mappings, 
            Dictionary<int,double> instanceID_Score_Mappings)
        {
            List<TableColumn> tgtCols =new List<TableColumn>();
            TableColumn col1=new TableColumn();
            col1.ColumnName = this._Instruction.IDFieldName;
            col1.AllowNull = false;
            col1.DbType = SqlDbType.Int;
            col1.IsFK = false;
            col1.IsIdentity = false;
            col1.IsIndexed = false;
            col1.IsPK = false;
            col1.Ordinal = 0;
            col1.Precision = 0;
            col1.Scale = 0;
            col1.Size = 0;
            tgtCols.Add(col1);

            TableColumn col2 = new TableColumn();
            col2.ColumnName = "CollapsedID";
            col2.AllowNull = false;
            col2.DbType = SqlDbType.Int;
            col2.IsFK = false;
            col2.IsIdentity = false;
            col2.IsIndexed = false;
            col2.IsPK = false;
            col2.Ordinal = 1;
            col2.Precision = 0;
            col2.Scale = 0;
            col2.Size = 0;
            tgtCols.Add(col2);

            TableColumn col3 = new TableColumn();
            col3.ColumnName = "SimilarityScore";
            col3.AllowNull = false;
            col3.DbType = SqlDbType.Float;
            col3.IsFK = false;
            col3.IsIdentity = false;
            col3.IsIndexed = false;
            col3.IsPK = false;
            col3.Ordinal = 2;
            col3.Precision = 0;
            col3.Scale = 0;
            col3.Size = 0;
            tgtCols.Add(col3);

            TableHelper.CreateTable(connStr, outputTableName, tgtCols.ToArray(), true);

            
            string spName = "insert_" + outputTableName + "_SP";
            string spBody = StoredProcUtil.CreateStoredProcedureStatementForInsert(DataProviderType.MSSQL, spName, string.Empty,
                                                                   outputTableName, tgtCols.ToArray());
            DbUtil.NewStoredProc(connStr, spName, spBody);

            SqlConnection conn =new SqlConnection(connStr);
            try
            {
                conn.Open();
                SqlCommand cmd = conn.CreateCommand();
                cmd.CommandType = CommandType.StoredProcedure;
                cmd.CommandText = spName;
                SqlParameter paramEntityID = new SqlParameter("@"+this._Instruction.IDFieldName, DBNull.Value);
                SqlParameter paramCollapseID = new SqlParameter("@CollapsedID", DBNull.Value);
                SqlParameter paramSimilarityScore = new SqlParameter("@SimilarityScore", DBNull.Value);

                int entityProcessed = 0;
                foreach (int entityID in instanceID_Score_Mappings.Keys)
                {
                    paramEntityID.Value = entityID;
                    paramCollapseID.Value = instanceID_CollapseID_Mappings[entityID];
                    if (instanceID_Score_Mappings.ContainsKey(entityID))
                        paramSimilarityScore.Value = instanceID_Score_Mappings[entityID];
                    else
                        paramSimilarityScore.Value = 0;
                    cmd.Parameters.Clear();
                    cmd.Parameters.AddRange(new SqlParameter[] {paramEntityID, paramCollapseID, paramSimilarityScore});
                    cmd.ExecuteNonQuery();

                    entityProcessed++;
                    if(this.WorkerStatusChangedEvent !=null && entityProcessed % 100 ==0 )
                    {
                        string msg = string.Format("Saving collapse mappings... processed {0} of {1}", entityProcessed,
                                                   instanceID_CollapseID_Mappings.Count);
                        int percent = entityProcessed*100/instanceID_CollapseID_Mappings.Count;
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName, msg, percent);
                    }
                }
            }
            catch (Exception ex)
            {
                if(this.WorkerErredEvent !=null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                conn.Close();
            }

            DbUtil.AddIndex(connStr, outputTableName, "CollapsedID", "idx_" + outputTableName + "_CollapseID");
            DbUtil.AddIndex(connStr, outputTableName, this._Instruction.IDFieldName, 
                "idx_" + outputTableName + "_" + this._Instruction.IDFieldName);
        }

        private void SaveCollapsedResult(
            string connStr,
            string collapsedTableName,
            DataTable dtEntities,
            Dictionary<int, int> instanceID_CollapseID_Mappings,
            List<string> descriptiveFieldNames,
            Dictionary<string,SqlDbType> fieldDbTypes,
            Dictionary<string,int> fieldSizes, 
            Dictionary<string,IAttributeComparer> fieldComparers,
            Dictionary<string,string> fieldValueDistTableNames)
        {
            List<TableColumn> tgtCols=new List<TableColumn>();
            TableColumn col1 = new TableColumn();
            col1.ColumnName = "CollapsedID";
            col1.AllowNull = false;
            col1.DbType = SqlDbType.Int;
            col1.IsFK = false;
            col1.IsIdentity = false;
            col1.IsIndexed = false;
            col1.IsPK = false;
            col1.Ordinal = 0;
            col1.Precision = 0;
            col1.Scale = 0;
            col1.Size = 0;
            tgtCols.Add(col1);

            foreach(string fieldName in descriptiveFieldNames)
            {
                TableColumn col=new TableColumn();
                col.AllowNull = true;
                col.ColumnName = fieldName;
                col.DbType = fieldDbTypes[fieldName];
                col.IsFK = false;
                col.IsIdentity = false;
                col.IsIndexed = false;
                col.IsPK = false;
                col.Ordinal = tgtCols.Count;
                col.Precision = 0;
                col.Scale = 0;
                col.Size = fieldSizes[fieldName];
                tgtCols.Add(col);
            }

            TableHelper.CreateTable(connStr,collapsedTableName,tgtCols.ToArray(),true);
            string spName = "insert_" + collapsedTableName + "_SP";
            string spBody = StoredProcUtil.CreateStoredProcedureStatementForInsert(
                DataProviderType.MSSQL, spName, string.Empty, collapsedTableName, tgtCols.ToArray());
            DbUtil.NewStoredProc(connStr, spName, spBody);

            if(this.WorkerStatusChangedEvent !=null)
            {
                this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName,
                                              "Saving collapse result...", 5);
            }
            Dictionary<int,List<int>> collapseID_EntityIDs=new Dictionary<int, List<int>>();
            foreach(int entityID in instanceID_CollapseID_Mappings.Keys)
            {
                int collapseID = instanceID_CollapseID_Mappings[entityID];
                if(collapseID_EntityIDs.ContainsKey(collapseID))
                {
                    List<int> entityIDs = collapseID_EntityIDs[collapseID];
                    entityIDs.Add(entityID);
                    collapseID_EntityIDs[collapseID] = entityIDs;
                }
                else
                {
                    List<int> entityIDs=new List<int>();
                    entityIDs.Add(entityID);
                    collapseID_EntityIDs.Add(collapseID, entityIDs);
                }
            }

            Dictionary<string,SqlParameter> parameters=new Dictionary<string, SqlParameter>();
            SqlParameter collapseIDParam = new SqlParameter("@CollapsedID", DBNull.Value);

            foreach(string fieldName in descriptiveFieldNames)
            {
                SqlParameter parameter=new SqlParameter("@"+fieldName,DBNull.Value);
                parameters.Add(fieldName, parameter);
            }

            int idColIndex = 0;
            for(int i=0;i<dtEntities.Columns.Count;i++)
            {
                if(dtEntities.Columns[i].ColumnName.ToLower()==this._Instruction.IDFieldName.ToLower())
                {
                    idColIndex = i;
                    break;
                }
            }
            Dictionary<int,int> instanceID_RowNumber=new Dictionary<int, int>();
            for(int i=0;i<dtEntities.Rows.Count;i++)
            {
                int instanceID = int.Parse(dtEntities.Rows[i][idColIndex].ToString());
                instanceID_RowNumber.Add(instanceID, i);
            }

            SqlConnection conn=new SqlConnection(connStr);
            try
            {
                conn.Open();
                SqlCommand cmd = conn.CreateCommand();
                cmd.CommandType = CommandType.StoredProcedure;
                cmd.CommandText = spName;

                int[] collapseIDs=new int[collapseID_EntityIDs.Count];
                collapseID_EntityIDs.Keys.CopyTo(collapseIDs, 0);
                for(int i=0;i<collapseIDs.Length;i++)
                {
                    cmd.Parameters.Clear();
                    collapseIDParam.Value = collapseIDs[i];
                    cmd.Parameters.Add(collapseIDParam);

                    List<int> entityIDs = collapseID_EntityIDs[collapseIDs[i]];
                    foreach(string fieldName in descriptiveFieldNames)
                    {
                        List<object> fieldValues=new List<object>();
                        for(int k=0;k<entityIDs.Count;k++)
                        {
                            int rowNum = instanceID_RowNumber[entityIDs[k]];
                            fieldValues.Add(dtEntities.Rows[rowNum][fieldName]);
                        }
                        IAttributeComparer comparer = fieldComparers[fieldName];
                        object mergedValue = null;
                        string distTblName = string.Empty;
                        if (fieldValueDistTableNames.ContainsKey(fieldName))
                            distTblName = fieldValueDistTableNames[fieldName];
                        mergedValue = comparer.Merge(fieldValues, conn, distTblName);
                        SqlParameter parameter = parameters[fieldName];
                        if (mergedValue == null)
                            parameter.Value = DBNull.Value;
                        else
                            parameter.Value = mergedValue;
                        cmd.Parameters.Add(parameter);
                    }
                    cmd.ExecuteNonQuery();

                    if(this.WorkerStatusChangedEvent !=null && (i+1)%100==0)
                    {
                        string msg = string.Format("Saving collapse result... {0} of {1} processed", (i + 1),
                                                   collapseIDs.Length);
                        int percent = (i + 1)*100/collapseIDs.Length;
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName, msg, percent);
                    }
                }
            }
            catch (Exception ex)
            {
                if(this.WorkerErredEvent !=null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                conn.Close();
            }
        }

        private void UpdateSimilarityScores(
            Dictionary<int,int> entityID_CollapseID_Mappings,
            DataTable dtEntities,
            Dictionary<string,IAttributeComparer> fieldComparers,
            Dictionary<string, string> fieldDistTableNames,
            ref Dictionary<int,double> entityID_Score)
        {
            Dictionary<int,List<int>> collapseID_EntityIDs_Mappings=new Dictionary<int, List<int>>();
            foreach(int entityID in entityID_CollapseID_Mappings.Keys)
            {
                int collapseID = entityID_CollapseID_Mappings[entityID];
                if(collapseID_EntityIDs_Mappings.ContainsKey(collapseID))
                {
                    List<int> entityIDs = collapseID_EntityIDs_Mappings[collapseID];
                    entityIDs.Add(entityID);
                    collapseID_EntityIDs_Mappings[collapseID] = entityIDs;
                }
                else
                {
                    List<int> entityIDs=new List<int>();
                    entityIDs.Add(entityID);
                    collapseID_EntityIDs_Mappings.Add(collapseID,entityIDs);
                }
            }
            int[] collapseIDs=new int[collapseID_EntityIDs_Mappings.Count];
            collapseID_EntityIDs_Mappings.Keys.CopyTo(collapseIDs, 0);

            Dictionary<int,int> entityID_RowNum =new Dictionary<int, int>();
            for(int i=0;i<dtEntities.Rows.Count;i++)
            {
                int entityID = int.Parse(dtEntities.Rows[i][this._Instruction.IDFieldName].ToString());
                entityID_RowNum.Add(entityID, i);
            }

            SqlConnection conn=new SqlConnection(this._Instruction.ConnStr);
            SqlCommand cmd = conn.CreateCommand();

            try
            {
                conn.Open();
                for(int i=0;i<collapseIDs.Length;i++)
                {
                    List<int> entityIDs = collapseID_EntityIDs_Mappings[collapseIDs[i]];
                    foreach(int entityID in entityIDs)
                    {
                        entityID_Score[entityID] = 0.0;
                    }
                    for (int j = 0; j < entityIDs.Count; j++)
                    {
                        double entityScore = 0.0;
                        DataRow dr1 = dtEntities.Rows[entityID_RowNum[entityIDs[j]]];
                        for (int k = j + 1; k < entityIDs.Count; k++)
                        {
                            DataRow dr2 = dtEntities.Rows[entityID_RowNum[entityIDs[k]]];
                            double overallScore = 0.0;
                            foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                            {
                                string swappedFieldName = string.Empty;
                                double fieldScore = this.GetFieldScoreConsideringSwaps(
                                    ref swappedFieldName, conn, fieldName,
                                    dr1, dr2, fieldComparers, fieldDistTableNames);
                                overallScore += 
                                    fieldScore * this._Instruction.DescriptiveAttributes[fieldName];
                                //if (fieldComparers.ContainsKey(fieldName))
                                //{
                                //    object fieldValue = dr1[fieldName];
                                //    object mergedValue = dr2[fieldName];
                                //    IAttributeComparer fieldComparer = fieldComparers[fieldName];
                                //    string unevenDistTblName = string.Empty;
                                //    if (fieldDistTableNames.ContainsKey(fieldName))
                                //        unevenDistTblName = fieldDistTableNames[fieldName];
                                //    double fieldScore =
                                //        fieldComparer.Compare(fieldValue,
                                //                              mergedValue,
                                //                              fieldPenalties[fieldName],
                                //                              conn,
                                //                              unevenDistTblName);
                                //    overallScore += fieldScore*this._Instruction.DescriptiveAttributes[fieldName];
                                //}
                            }
                            if(overallScore >= this._Instruction.AssociateThreshold)
                            {
                                foreach(string fieldName in this._Instruction.AssociativeAttributes.Keys)
                                {
                                    string swappedFieldName = string.Empty;
                                    double fieldScore = this.GetFieldScoreConsideringSwaps(
                                        ref swappedFieldName, conn, fieldName,
                                        dr1, dr2, fieldComparers, fieldDistTableNames);
                                    overallScore +=
                                        fieldScore * this._Instruction.AssociativeAttributes[fieldName];
                                    //if (fieldComparers.ContainsKey(fieldName))
                                    //{
                                    //    object fieldValue = dr1[fieldName];
                                    //    object mergedValue = dr2[fieldName];
                                    //    IAttributeComparer fieldComparer = fieldComparers[fieldName];
                                    //    string unevenDistTblName = string.Empty;
                                    //    if (fieldDistTableNames.ContainsKey(fieldName))
                                    //        unevenDistTblName = fieldDistTableNames[fieldName];
                                    //    double fieldScore =
                                    //        fieldComparer.Compare(fieldValue,
                                    //                              mergedValue,
                                    //                              fieldPenalties[fieldName],
                                    //                              conn,
                                    //                              unevenDistTblName);
                                    //    overallScore += fieldScore * this._Instruction.AssociativeAttributes[fieldName];
                                    //}
                                }
                            }
                            if (overallScore > entityScore)
                                entityScore = overallScore;
                            if (overallScore > entityID_Score[entityIDs[k]])
                                entityID_Score[entityIDs[k]] = overallScore;
                        }
                        if (entityID_Score[entityIDs[j]] < entityScore)
                            entityID_Score[entityIDs[j]] = entityScore;
                    }

                    foreach(int entityID in entityIDs)
                    {
                        cmd.CommandText =
                            string.Format("update {0} set {1}={2} where {3}={4}",
                                          this._Instruction.OutputTableName,
                                          "SimilarityScore",
                                          entityID_Score[entityID],
                                          this._Instruction.IDFieldName,
                                          entityID);
                        cmd.ExecuteNonQuery();
                    }

                    if (this.WorkerStatusChangedEvent != null && (i + 1) % 10 == 0)
                    {
                        string msg = string.Format("Updating similarity score... processed {0} of {1}", (i + 1),
                                                   collapseIDs.Length);
                        int percent = (i + 1) * 100 / collapseIDs.Length;
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName, msg, percent);
                    }
                }
            }
            catch (Exception ex)
            {
                if(this.WorkerErredEvent !=null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
            finally
            {
                conn.Close();
            }
        }

        private void SaveUnderCollapsed(DataTable dtUnderCollapsedEntities, string excelFilePath)
        {
            List<int> underCollapsedIDs = new List<int>();
            if (dtUnderCollapsedEntities != null && dtUnderCollapsedEntities.Rows.Count > 0)
            {
                foreach (DataRow dr in dtUnderCollapsedEntities.Rows)
                {

                    int collapseID1 = int.Parse(dr[CollapseIDFieldName].ToString());
                    if (!underCollapsedIDs.Contains(collapseID1))
                    {
                        underCollapsedIDs.Add(collapseID1);
                    }
                    int collapseID2 = int.Parse(dr[CollapseIDFieldName + "_Compare"].ToString());
                    if (!underCollapsedIDs.Contains(collapseID2))
                    {
                        underCollapsedIDs.Add(collapseID2);
                    }
                }
            }

            if (underCollapsedIDs.Count > 0)
            {
                try
                {
                    if(this.WorkerStatusChangedEvent !=null)
                    {
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName,
                                                      "Saving under collapsed entities to excel file...", 90);
                    }

                    string collapsedIDFieldName = "CollapsedID";
                    List<string> colHeaders = new List<string>();
                    colHeaders.Add(collapsedIDFieldName);
                    colHeaders.Add(this._Instruction.IDFieldName);
                    colHeaders.Add("SimilarityScore");
                    foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                    {
                        colHeaders.Add(fieldName);
                    }
                    foreach (string fieldName in this._Instruction.AssociativeAttributes.Keys)
                    {
                        colHeaders.Add(fieldName);
                    }

                    // master 
                    DataTable dtMaster = DbUtil.GrabDataByKeyValues(
                        this._Instruction.ConnStr,
                        this._Instruction.CollapsedTableName,
                        collapsedIDFieldName,
                        underCollapsedIDs);
                    Dictionary<string, string> masterField_ColHeader_Mappings = new Dictionary<string, string>();
                    foreach (DataColumn col in dtMaster.Columns)
                    {
                        if (colHeaders.Contains(col.ColumnName))
                        {
                            masterField_ColHeader_Mappings.Add(col.ColumnName, col.ColumnName);
                        }
                    }

                    // association 
                    DataTable dtMappings = DbUtil.GrabDataByKeyValues(
                        this._Instruction.ConnStr,
                        this._Instruction.OutputTableName,
                        collapsedIDFieldName,
                        underCollapsedIDs);
                    List<int> entityIDs = new List<int>();
                    Dictionary<int, List<int>> collapseID_EntityID_Mappings = new Dictionary<int, List<int>>();
                    foreach (DataRow dr in dtMappings.Rows)
                    {
                        int entityID = int.Parse(dr[this._Instruction.IDFieldName].ToString());
                        if (!entityIDs.Contains(entityID))
                        {
                            entityIDs.Add(entityID);
                        }
                        int collapseID = int.Parse(dr[collapsedIDFieldName].ToString());
                        if (!collapseID_EntityID_Mappings.ContainsKey(collapseID))
                        {
                            List<int> detailIDs = new List<int>();
                            detailIDs.Add(entityID);
                            collapseID_EntityID_Mappings.Add(collapseID, detailIDs);
                        }
                        else
                        {
                            List<int> detailIDs = collapseID_EntityID_Mappings[collapseID];
                            if (!detailIDs.Contains(entityID))
                            {
                                detailIDs.Add(entityID);
                            }
                            collapseID_EntityID_Mappings[collapseID] = detailIDs;
                        }
                    }

                    // detail 
                    string selectColNameSql = "select a." + this._Instruction.IDFieldName + ",";
                    foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                    {
                        selectColNameSql += "a." + fieldName + ",";
                    }
                    foreach (string fieldName in this._Instruction.AssociativeAttributes.Keys)
                    {
                        selectColNameSql += "a." + fieldName + ",";
                    }
                    string selectSql =
                        string.Format(
                            selectColNameSql +
                            " b.SimilarityScore from {0} a, {1} b where a.{2}=b.{2}",
                            (string.IsNullOrEmpty(this._Instruction.SchemaName)
                                 ? ""
                                 : this._Instruction.SchemaName + ".") +
                            this._Instruction.TableName,
                            this._Instruction.OutputTableName,
                            this._Instruction.IDFieldName);
                    DataTable dtDetail = DbUtil.GrabDataBySqlAndKeyValues(
                        this._Instruction.ConnStr,
                        selectSql,
                        this._Instruction.IDFieldName,
                        entityIDs);

                    Dictionary<string, string> detailField_ColHeader_Mappings = new Dictionary<string, string>();
                    foreach (DataColumn col in dtDetail.Columns)
                    {
                        if (colHeaders.Contains(col.ColumnName))
                        {
                            detailField_ColHeader_Mappings.Add(col.ColumnName, col.ColumnName);
                        }
                    }

                    ExcelTableWithGroupsUtil.OnPopulatingRow += new PopulatingRow(ExcelTableWithGroupsUtil_OnPopulatingRow);

                    ExcelTableWithGroupsUtil.FillTableWithGroup(
                        excelFilePath, colHeaders, detailField_ColHeader_Mappings,
                        masterField_ColHeader_Mappings, collapsedIDFieldName,
                        this._Instruction.IDFieldName, collapseID_EntityID_Mappings,
                        dtMaster, dtDetail, false);

                    ExcelTableWithGroupsUtil.OnPopulatingRow -= ExcelTableWithGroupsUtil_OnPopulatingRow;
                }
                catch (Exception ex)
                {
                    if(this.WorkerErredEvent !=null)
                    {
                        this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                    }
                }
            }
        }

        private void SaveOverCollapsed(DataTable dtOverCollapsedEntities, string excelFilePath)
        {
            List<int> overCollapsedIDs = new List<int>();
            if (dtOverCollapsedEntities != null && dtOverCollapsedEntities.Rows.Count > 0)
            {
                foreach (DataRow dr in dtOverCollapsedEntities.Rows)
                {

                    int collapseID1 = int.Parse(dr[CollapseIDFieldName].ToString());
                    if (!overCollapsedIDs.Contains(collapseID1))
                    {
                        overCollapsedIDs.Add(collapseID1);
                    }
                    int collapseID2 = int.Parse(dr[CollapseIDFieldName + "_Compare"].ToString());
                    if (!overCollapsedIDs.Contains(collapseID2))
                    {
                        overCollapsedIDs.Add(collapseID2);
                    }
                }
            }

            if (overCollapsedIDs.Count > 0)
            {
                try
                {
                    if (this.WorkerStatusChangedEvent != null)
                    {
                        this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName,
                                                      "Saving over collapsed entities to excel file...", 95);
                    }

                    // column headers
                    string collapsedIDFieldName = "CollapsedID";
                    List<string> colHeaders = new List<string>();
                    colHeaders.Add(collapsedIDFieldName);
                    colHeaders.Add(this._Instruction.IDFieldName);
                    colHeaders.Add("SimilarityScore");
                    foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                    {
                        colHeaders.Add(fieldName);
                    }
                    foreach (string fieldName in this._Instruction.AssociativeAttributes.Keys)
                    {
                        colHeaders.Add(fieldName);
                    }

                    // master
                    DataTable dtMaster = DbUtil.GrabDataByKeyValues(
                        this._Instruction.ConnStr,
                        this._Instruction.CollapsedTableName,
                        collapsedIDFieldName,
                        overCollapsedIDs);
                    Dictionary<string, string> masterField_ColHeader_Mappings = new Dictionary<string, string>();
                    foreach (DataColumn col in dtMaster.Columns)
                    {
                        if (colHeaders.Contains(col.ColumnName))
                        {
                            masterField_ColHeader_Mappings.Add(col.ColumnName, col.ColumnName);
                        }
                    }

                    // association
                    DataTable dtMappings = DbUtil.GrabDataByKeyValues(
                        this._Instruction.ConnStr,
                        this._Instruction.OutputTableName,
                        collapsedIDFieldName,
                        overCollapsedIDs);
                    List<int> entityIDs = new List<int>();
                    Dictionary<int, List<int>> collapseID_EntityID_Mappings = new Dictionary<int, List<int>>();
                    foreach (DataRow dr in dtMappings.Rows)
                    {
                        int entityID = int.Parse(dr[this._Instruction.IDFieldName].ToString());
                        if (!entityIDs.Contains(entityID))
                        {
                            entityIDs.Add(entityID);
                        }
                        int collapseID = int.Parse(dr[collapsedIDFieldName].ToString());
                        if (!collapseID_EntityID_Mappings.ContainsKey(collapseID))
                        {
                            List<int> detailIDs = new List<int>();
                            detailIDs.Add(entityID);
                            collapseID_EntityID_Mappings.Add(collapseID, detailIDs);
                        }
                        else
                        {
                            List<int> detailIDs = collapseID_EntityID_Mappings[collapseID];
                            if (!detailIDs.Contains(entityID))
                            {
                                detailIDs.Add(entityID);
                            }
                            collapseID_EntityID_Mappings[collapseID] = detailIDs;
                        }
                    }

                    // detail
                    string selectColNameSql = "select a." + this._Instruction.IDFieldName + ",";
                    foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                    {
                        selectColNameSql += "a." + fieldName + ",";
                    }
                    foreach (string fieldName in this._Instruction.AssociativeAttributes.Keys)
                    {
                        selectColNameSql += "a." + fieldName + ",";
                    }
                    string selectSql =
                        string.Format(
                            selectColNameSql + " b.SimilarityScore from {0} a, {1} b where a.{2}=b.{2}",
                            (string.IsNullOrEmpty(this._Instruction.SchemaName)
                                 ? ""
                                 : this._Instruction.SchemaName + ".") + this._Instruction.TableName,
                            this._Instruction.OutputTableName,
                            this._Instruction.IDFieldName);
                    DataTable dtDetail = DbUtil.GrabDataBySqlAndKeyValues(
                        this._Instruction.ConnStr,
                        selectSql,
                        this._Instruction.IDFieldName,
                        entityIDs);
                    Dictionary<string, string> detailField_ColHeader_Mappings = new Dictionary<string, string>();
                    foreach (DataColumn col in dtDetail.Columns)
                    {
                        if (colHeaders.Contains(col.ColumnName))
                        {
                            detailField_ColHeader_Mappings.Add(col.ColumnName, col.ColumnName);
                        }
                    }

                    ExcelTableWithGroupsUtil.OnPopulatingRow += new PopulatingRow(ExcelTableWithGroupsUtil_OnPopulatingRow);

                    ExcelTableWithGroupsUtil.FillTableWithGroup(
                        excelFilePath, colHeaders,
                        detailField_ColHeader_Mappings,
                        masterField_ColHeader_Mappings,
                        collapsedIDFieldName,
                        this._Instruction.IDFieldName,
                        collapseID_EntityID_Mappings,
                        dtMaster, dtDetail, false);

                    ExcelTableWithGroupsUtil.OnPopulatingRow -= ExcelTableWithGroupsUtil_OnPopulatingRow;
                }
                catch (Exception ex)
                {
                    if(this.WorkerErredEvent !=null)
                    {
                        this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                    }
                }
            }
        }

        private void SaveFinalResults(
            string excelFilePath,
            string collapseIDFieldName,
            string scoreFieldName)
        {
            try
            {
                if (this.WorkerStatusChangedEvent != null)
                {
                    this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName,
                                                  "Saving final collapsed results to excel file...", 99);
                }

                // column headers
                List<string> colHeaders = new List<string>();
                colHeaders.Add(collapseIDFieldName);
                colHeaders.Add(this._Instruction.IDFieldName);
                colHeaders.Add(scoreFieldName);
                foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                {
                    colHeaders.Add(fieldName);
                }
                foreach (string fieldName in this._Instruction.AssociativeAttributes.Keys)
                {
                    colHeaders.Add(fieldName);
                }

                // master
                DataTable dtMaster =
                    DbUtil.GetTableData(
                        this._Instruction.ConnStr,
                        this._Instruction.CollapsedTableName);
                Dictionary<string, string> masterField_ColHeader_Mappings = new Dictionary<string, string>();
                foreach (DataColumn col in dtMaster.Columns)
                {
                    if (colHeaders.Contains(col.ColumnName))
                    {
                        masterField_ColHeader_Mappings.Add(col.ColumnName, col.ColumnName);
                    }
                }

                // detail
                string selectColNameSql = "select a." + this._Instruction.IDFieldName + ",";
                foreach (string fieldName in this._Instruction.DescriptiveAttributes.Keys)
                {
                    selectColNameSql += "a."+fieldName + ",";
                }
                foreach (string fieldName in this._Instruction.AssociativeAttributes.Keys)
                {
                    selectColNameSql += "a." + fieldName + ",";
                }
                string selectSql =
                    string.Format(
                        selectColNameSql +
                        " b.{0}, b.{1} from {2} a, {3} b where a.{4}=b.{4}",
                        collapseIDFieldName,
                        scoreFieldName,
                        (string.IsNullOrEmpty(this._Instruction.SchemaName)
                             ? ""
                             : this._Instruction.SchemaName + ".") + this._Instruction.TableName,
                        this._Instruction.OutputTableName,
                        this._Instruction.IDFieldName);
                DataTable dtDetail = DbUtil.GetTableDataUsingSelectStatement(
                    this._Instruction.ConnStr, selectSql);
                Dictionary<string, string> detailField_ColHeader_Mappings = new Dictionary<string, string>();
                foreach (DataColumn col in dtDetail.Columns)
                {
                    if (colHeaders.Contains(col.ColumnName))
                    {
                        detailField_ColHeader_Mappings.Add(col.ColumnName, col.ColumnName);
                    }
                }

                ExcelTableWithGroupsUtil.OnPopulatingRow += new PopulatingRow(ExcelTableWithGroupsUtil_OnPopulatingRow);

                ExcelTableWithGroupsUtil.FillTableWithGroup(
                    excelFilePath, colHeaders,
                    detailField_ColHeader_Mappings,
                    masterField_ColHeader_Mappings,
                    collapseIDFieldName,
                    collapseIDFieldName,
                    dtMaster, dtDetail, false);

                ExcelTableWithGroupsUtil.OnPopulatingRow -= ExcelTableWithGroupsUtil_OnPopulatingRow;
            }
            catch (Exception ex)
            {
                if (this.WorkerErredEvent != null)
                {
                    this.WorkerErredEvent(this._JobTicketID, this.GetType(), this._Instruction, ex.ToString(),true);
                }
            }
        }

        void ExcelTableWithGroupsUtil_OnPopulatingRow(int rowNumber, int totalRows)
        {
            if(this.WorkerStatusChangedEvent !=null)
            {
                int percent = (rowNumber*100)/totalRows;
                percent = percent%100;
                this.WorkerStatusChangedEvent(this._JobTicketID, this.GetType(), this._Instruction, this.WorkflowName, this.ComponentName,
                                              "Populating excel file " + rowNumber + " of " + totalRows, percent);
            }
        }

        #endregion

        #region swap
        private double GetFieldScoreConsideringSwaps(
            ref string swappableFieldCandidate,
            SqlConnection conn, 
            string currentFieldName, 
            DataRow dr1, DataRow dr2, 
            Dictionary<string,IAttributeComparer> fieldComparers,
            Dictionary<string,string> fieldValueDistTableNames)
        {
            string distTblName = string.Empty;
            if (fieldValueDistTableNames.ContainsKey(currentFieldName))
                distTblName = fieldValueDistTableNames[currentFieldName];
            object fieldValue1a = dr1[currentFieldName];
            object fieldValue2a = dr2[currentFieldName];
            double fieldScore1 = fieldComparers[currentFieldName].Compare(
                fieldValue1a, fieldValue2a,
                this._Instruction.PenaltiesWhenNotMatched[currentFieldName], 
                conn, distTblName);
            if (fieldScore1 <= 0 && fieldValue1a != null && fieldValue2a != null &&
                fieldValue1a.ToString().Length > 0 && fieldValue2a.ToString().Length > 0 &&
                this._Instruction.SwappedFieldsCollection != null && 
                this._Instruction.SwappedFieldsCollection.Count > 0)
            {
                for (int i = 0; i < this._Instruction.SwappedFieldsCollection.Count; i++)
                {
                    List<string> swappableFields = this._Instruction.SwappedFieldsCollection[i];
                    if (swappableFields.Contains(currentFieldName))
                    {
                        foreach (string swappableField in swappableFields)
                        {
                            if (swappableField != currentFieldName)
                            {
                                double fieldScore2a = fieldComparers[currentFieldName].Compare(
                                    dr1[currentFieldName], dr2[swappableField],
                                    this._Instruction.PenaltiesWhenNotMatched[currentFieldName],
                                    conn, distTblName);

                                if (fieldScore2a > fieldScore1)
                                {
                                    fieldScore1 = fieldScore2a;
                                    swappableFieldCandidate = swappableField;
                                }
                            }
                        }
                    }
                }
            }
            return fieldScore1;
        }
        #endregion
    }
}
